use super::index::{FieldId, SegmentIndex};
pub(crate) enum PhraseReadError<Control> {
Storage(super::sealed::SealedSegmentError),
Control(Control),
}
impl<Control> From<super::postings::PostingsError> for PhraseReadError<Control> {
fn from(error: super::postings::PostingsError) -> Self {
Self::Storage(error.into())
}
}
impl<Control> From<super::sealed::SealedSegmentError> for PhraseReadError<Control> {
fn from(error: super::sealed::SealedSegmentError) -> Self {
Self::Storage(error)
}
}
struct SegmentPhrase<'index> {
readers: Vec<Option<super::postings::PostingsReader<'index>>>,
order: Vec<usize>,
rows: u32,
}
pub(crate) struct PreparedPhrase<'index> {
segments: Vec<SegmentPhrase<'index>>,
slop: u32,
bytes: usize,
}
impl<'index> PreparedPhrase<'index> {
pub(crate) fn allocation_bytes(&self) -> usize {
self.bytes
}
pub(crate) fn new<Control>(
index: &'index super::index::LexicalIndex,
query: &super::query::LexicalQuery,
mut reserve: impl FnMut(Option<usize>) -> Result<(), Control>,
mut checkpoint: impl FnMut() -> Result<(), Control>,
) -> Result<Option<Self>, PhraseReadError<Control>> {
let super::query::LexicalQuery::Phrase { terms, slop, field } = query else {
return Ok(None);
};
let mut work =
super::control::WorkCheck::new(|| checkpoint().map_err(PhraseReadError::Control));
work.check_now()?;
let mut bytes = index
.segments()
.len()
.checked_mul(std::mem::size_of::<SegmentPhrase<'_>>());
for segment in index.segments() {
work.step()?;
bytes = bytes.and_then(|bytes| {
terms
.len()
.checked_mul(
std::mem::size_of::<Option<super::postings::PostingsReader<'_>>>()
+ std::mem::size_of::<usize>(),
)
.and_then(|extra| bytes.checked_add(extra))
});
for term in terms {
work.step()?;
let extra = segment.position_reader_bytes_controlled(term, *field, &mut work)?;
bytes = bytes.and_then(|bytes| extra.and_then(|extra| bytes.checked_add(extra)));
}
}
reserve(bytes).map_err(PhraseReadError::Control)?;
let bytes = bytes.ok_or(PhraseReadError::Storage(
super::sealed::SealedSegmentError::Geometry("phrase allocation size overflow"),
))?;
let mut segments = Vec::with_capacity(index.segments().len());
for segment in index.segments() {
work.step()?;
let mut readers = Vec::with_capacity(terms.len());
for term in terms {
work.step()?;
readers.push(segment.position_reader_controlled(term, *field, &mut work)?);
}
let mut order = Vec::with_capacity(terms.len());
for slot in 0..terms.len() {
work.step()?;
order.push(slot);
}
let frequency = |slot: &usize| {
readers
.get(*slot)
.and_then(Option::as_ref)
.map_or(0, |reader| reader.document_frequency())
};
super::control::sort_by(
&mut order,
|left, right| frequency(left).cmp(&frequency(right)),
&mut work,
)?;
segments.push(SegmentPhrase {
readers,
order,
rows: segment.row_count(),
});
}
work.check_now()?;
Ok(Some(Self {
segments,
slop: *slop,
bytes,
}))
}
pub(crate) fn matches<Control>(
&self,
doc: super::search::GlobalDocId,
mut reserve: impl FnMut(Option<usize>) -> Result<(), Control>,
mut checkpoint: impl FnMut() -> Result<(), Control>,
) -> Result<bool, PhraseReadError<Control>> {
let invalid = || {
PhraseReadError::Storage(super::sealed::SealedSegmentError::Geometry(
"phrase candidate is outside its pinned segment",
))
};
let mut work =
super::control::WorkCheck::new(|| checkpoint().map_err(PhraseReadError::Control));
work.check_now()?;
let segment = self
.segments
.get(doc.segment as usize)
.ok_or_else(invalid)?;
if doc.row >= segment.rows {
return Err(invalid());
}
let views_bytes = segment
.readers
.len()
.checked_mul(std::mem::size_of::<Option<super::postings::RowPositions<'_>>>());
let scratch = views_bytes.and_then(|bytes| self.bytes.checked_add(bytes));
reserve(scratch).map_err(PhraseReadError::Control)?;
let mut views = vec![None; segment.readers.len()];
let mut positions = 0_usize;
let mut maximum = 1_usize;
for slot in &segment.order {
work.step()?;
let Some(reader) = segment.readers.get(*slot).and_then(Option::as_ref) else {
return Ok(false);
};
let Some(view) = reader.positions_controlled(doc.row, &mut work)? else {
return Ok(false);
};
positions = positions.checked_add(view.len()).ok_or_else(invalid)?;
maximum = maximum.max(view.len());
*views.get_mut(*slot).ok_or_else(invalid)? = Some(view);
}
let bytes = scratch
.and_then(|bytes| {
segment
.readers
.len()
.checked_mul(std::mem::size_of::<Vec<u32>>())
.and_then(|extra| bytes.checked_add(extra))
})
.and_then(|bytes| {
positions
.checked_mul(std::mem::size_of::<u32>())
.and_then(|extra| bytes.checked_add(extra))
})
.and_then(|bytes| {
maximum
.checked_mul(2 * std::mem::size_of::<(u32, u32)>())
.and_then(|extra| bytes.checked_add(extra))
});
reserve(bytes).map_err(PhraseReadError::Control)?;
let mut streams = Vec::with_capacity(views.len());
for view in views {
work.step()?;
streams.push(view.ok_or_else(invalid)?.decode_controlled(&mut work)?);
}
Ok(minimum_displacement_controlled(&streams, &mut work)?
.is_some_and(|cost| cost <= self.slop))
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PhraseQuery {
pub terms: Vec<Vec<u8>>,
pub slop: u32,
pub field: FieldId,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PhraseError {
Empty,
}
impl std::fmt::Display for PhraseError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("a phrase query needs at least one term")
}
}
impl std::error::Error for PhraseError {}
#[must_use]
pub fn minimum_displacement(streams: &[Vec<u32>]) -> Option<u32> {
let mut work = super::control::WorkCheck::new(|| Ok::<(), std::convert::Infallible>(()));
match minimum_displacement_controlled(streams, &mut work) {
Ok(displacement) => displacement,
Err(never) => match never {},
}
}
pub(crate) fn minimum_displacement_controlled<E>(
streams: &[Vec<u32>],
work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
) -> Result<Option<u32>, E> {
work.check_now()?;
let Some(first) = streams.first() else {
return Ok(None);
};
if streams.len() == 1 {
return Ok((!first.is_empty()).then_some(0));
}
for stream in streams {
work.step()?;
if stream.is_empty() {
return Ok(None);
}
}
let mut best: Option<u32> = None;
for anchor in first {
work.step()?;
let mut previous: Vec<(u32, u32)> = vec![(*anchor, 0)];
for (offset, stream) in streams.iter().enumerate().skip(1) {
#[cfg(test)]
DP_TERM_OFFSET.with(|value| value.set(offset));
work.step()?;
let expected = u32::try_from(offset).unwrap_or(u32::MAX);
let mut current: Vec<(u32, u32)> = Vec::with_capacity(stream.len());
for candidate in stream {
work.step()?;
let mut best_previous: Option<u32> = None;
for (position, cost) in &previous {
work.step()?;
#[cfg(test)]
DP_PREDECESSOR_PROBES.with(|value| value.set(value.get() + 1));
if position <= candidate {
best_previous = Some(best_previous.map_or(*cost, |best| best.min(*cost)));
}
}
let Some(best_previous) = best_previous else {
continue;
};
let relative = candidate.saturating_sub(*anchor);
let displacement = relative.abs_diff(expected);
current.push((*candidate, best_previous.saturating_add(displacement)));
}
if current.is_empty() {
previous.clear();
break;
}
previous = current;
}
for (_, cost) in previous {
work.step()?;
best = Some(best.map_or(cost, |current| current.min(cost)));
}
}
work.check_now()?;
Ok(best)
}
#[cfg(test)]
thread_local! {
static DP_PREDECESSOR_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
static DP_TERM_OFFSET: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
#[must_use]
pub fn streams_match(streams: &[Vec<u32>], slop: u32) -> bool {
minimum_displacement(streams).is_some_and(|cost| cost <= slop)
}
pub fn search_segment(
segment: &SegmentIndex,
query: &PhraseQuery,
) -> Result<Vec<u32>, PhraseError> {
if query.terms.is_empty() {
return Err(PhraseError::Empty);
}
let Some(first) = query.terms.first() else {
return Err(PhraseError::Empty);
};
let Some(anchor_list) = segment.posting_list(first, query.field) else {
return Ok(Vec::new());
};
let mut matches = Vec::new();
for posting in anchor_list.postings() {
let mut streams: Vec<Vec<u32>> = Vec::with_capacity(query.terms.len());
let mut complete = true;
for term in &query.terms {
let Some(list) = segment.posting_list(term, query.field) else {
complete = false;
break;
};
let Some(entry) = list
.postings()
.iter()
.find(|candidate| candidate.docid == posting.docid)
else {
complete = false;
break;
};
streams.push(entry.positions.clone());
}
if complete && streams_match(&streams, query.slop) {
matches.push(posting.docid);
}
}
Ok(matches)
}
#[cfg(test)]
#[allow(
clippy::expect_used,
clippy::indexing_slicing,
clippy::panic,
clippy::unwrap_used
)]
mod tests {
use super::*;
use crate::fts::index::{DEFAULT_FIELD, Document};
use crate::fts::tokenizer::{Analyzer, Profile, TokenizerConfig, Vocabulary};
#[test]
fn astra_18_prepared_phrase_cancels_during_position_decode() {
use crate::fts::index::LexicalIndex;
use crate::fts::postings::POSITION_DECODE_PROBES;
use crate::fts::query::LexicalQuery;
use crate::fts::sealed::SealedSegment;
use crate::fts::search::GlobalDocId;
let analyzer = Analyzer::new(TokenizerConfig::text_default()).expect("analyzer");
let mut active = SegmentIndex::new();
let text = format!("{}beta", "alpha ".repeat(4_096));
active
.push_document(&analyzer, &Document::with_text(&text))
.expect("long row");
let bytes = SealedSegment::seal(&active)
.expect("seal")
.encode_region()
.expect("persist");
let mut index = LexicalIndex::new();
index.push_sealed(SealedSegment::decode_region(&bytes).expect("reopen"));
let query =
LexicalQuery::phrase(vec![b"alpha".to_vec(), b"beta".to_vec()], 0, DEFAULT_FIELD);
let prepared = PreparedPhrase::new(&index, &query, |_| Ok::<_, &str>(()), || Ok(()))
.unwrap_or_else(|_| panic!("prepare"))
.expect("phrase");
POSITION_DECODE_PROBES.with(|value| value.set(0));
crate::fts::preparation_observer::begin();
let result = prepared.matches(
GlobalDocId { segment: 0, row: 0 },
|_| Ok(()),
|| {
if POSITION_DECODE_PROBES.with(std::cell::Cell::get) >= 64 {
Err("cancelled")
} else {
Ok(())
}
},
);
let probes = POSITION_DECODE_PROBES.with(std::cell::Cell::get);
let observed = crate::fts::preparation_observer::phrase_position_work();
crate::fts::preparation_observer::take();
println!("prepared_decode_probes={probes}, position_work={observed:?}");
assert!(matches!(result, Err(PhraseReadError::Control("cancelled"))));
assert!(
(64..=128).contains(&probes),
"prepared phrase must check inside row decode"
);
assert_eq!(
observed.0, probes,
"partial work survives the cancellation error"
);
assert!(
prepared
.matches(
GlobalDocId { segment: 0, row: 0 },
|_| Ok::<_, ()>(()),
|| Ok(())
)
.unwrap_or_else(|_| panic!("clean phrase")),
"last alpha and beta are adjacent"
);
}
#[test]
fn astra_18_phrase_cancels_inside_predecessor_scan() {
let streams = vec![vec![0], (1..=4_096).collect(), vec![4_097]];
assert_eq!(minimum_displacement(&streams), Some(4_095));
DP_PREDECESSOR_PROBES.with(|value| value.set(0));
DP_TERM_OFFSET.with(|value| value.set(0));
let mut checks = 0;
let mut work = super::super::control::WorkCheck::new(|| {
checks += 1;
if DP_TERM_OFFSET.with(std::cell::Cell::get) == 2
&& DP_PREDECESSOR_PROBES.with(std::cell::Cell::get) > 4_096
{
Err("cancelled")
} else {
Ok(())
}
});
let result = minimum_displacement_controlled(&streams, &mut work);
let probes = DP_PREDECESSOR_PROBES.with(std::cell::Cell::get);
println!("checks={checks}, predecessor_probes={probes}");
assert_eq!(result, Err("cancelled"));
assert!(
probes > 4_096 && probes <= 4_096 + 128,
"must stop within the final predecessor scan"
);
let mut work = super::super::control::WorkCheck::new(|| Ok::<(), ()>(()));
assert_eq!(
minimum_displacement_controlled(&streams, &mut work),
Ok(Some(4_095))
);
}
#[test]
fn astra_11_phrase_slop_repetition_and_field_boundaries_match_oracle() {
use crate::fts::index::LexicalIndex;
use crate::fts::query::LexicalQuery;
use crate::fts::sealed::SealedSegment;
use crate::fts::search::GlobalDocId;
let texts = [
"alpha beta alpha",
"alpha filler beta alpha",
"beta alpha",
"alpha alpha beta",
"alpha the beta",
];
let literal = [
[vec![0, 2], vec![1]],
[vec![0, 3], vec![2]],
[vec![1], vec![0]],
[vec![0, 1], vec![2]],
[vec![0], vec![2]],
];
let analyzer = Analyzer::new(TokenizerConfig::text_default()).expect("analyzer");
let mut active = SegmentIndex::new();
for text in texts {
let mut doc = Document::with_text(text);
doc.set(FieldId(1), "beta alpha");
active.push_document(&analyzer, &doc).expect("document");
}
let bytes = SealedSegment::seal(&active)
.expect("seal")
.encode_region()
.expect("persist");
let mut index = LexicalIndex::new();
index
.push_sealed(SealedSegment::decode_region(&bytes).expect("reopen persisted positions"));
fn enumerate(streams: &[Vec<u32>], chosen: &mut Vec<u32>, slop: u32) -> bool {
if chosen.len() == streams.len() {
let first = chosen[0];
return chosen
.iter()
.enumerate()
.map(|(i, value)| value.saturating_sub(first).abs_diff(i as u32))
.sum::<u32>()
<= slop;
}
for value in &streams[chosen.len()] {
if chosen.last().is_none_or(|last| value >= last) {
chosen.push(*value);
if enumerate(streams, chosen, slop) {
return true;
}
chosen.pop();
}
}
false
}
for field in [DEFAULT_FIELD, FieldId(1), FieldId(2)] {
for words in [vec![0, 1], vec![0, 0], vec![1, 0], vec![0, 1, 0]] {
let terms = words
.iter()
.map(|word| {
if *word == 0 {
b"alpha".to_vec()
} else {
b"beta".to_vec()
}
})
.collect::<Vec<_>>();
for slop in [0, 1, 2, 6] {
let query = LexicalQuery::phrase(terms.clone(), slop, field);
let prepared =
PreparedPhrase::new(&index, &query, |_| Ok::<_, ()>(()), || Ok(()))
.unwrap_or_else(|_| panic!("prepare"))
.expect("phrase");
for (row, source) in literal.iter().enumerate() {
let streams = words
.iter()
.map(|word| match field.0 {
0 => source[*word].clone(),
1 => vec![if *word == 0 { 1 } else { 0 }],
_ => vec![],
})
.collect::<Vec<_>>();
let expected = enumerate(&streams, &mut Vec::new(), slop);
let actual = prepared
.matches(
GlobalDocId {
segment: 0,
row: row as u32,
},
|_| Ok::<_, ()>(()),
|| Ok(()),
)
.unwrap_or_else(|_| panic!("positions"));
assert_eq!(
actual, expected,
"row={row} field={field:?} words={words:?} slop={slop}"
);
if field.0 < 2 {
assert_eq!(
actual,
crate::fts::query::phrase_matches(
&analyzer,
if field.0 == 0 {
texts[row]
} else {
"beta alpha"
},
&terms,
slop
),
"old text matcher secondary control"
);
}
}
}
}
}
}
fn segment_of(analyzer: &Analyzer, texts: &[&str]) -> SegmentIndex {
let mut segment = SegmentIndex::new();
for text in texts {
segment
.push_document(analyzer, &Document::with_text(text))
.expect("indexable");
}
segment
}
fn code_analyzer() -> Analyzer {
Analyzer::new(Profile::Code.config()).expect("valid config")
}
fn phrase(terms: &[&str], slop: u32) -> PhraseQuery {
PhraseQuery {
terms: terms.iter().map(|term| term.as_bytes().to_vec()).collect(),
slop,
field: DEFAULT_FIELD,
}
}
#[test]
fn slop_zero_and_slop_one_semantics_match_the_pinned_fixtures() {
let analyzer = code_analyzer();
let segment = segment_of(
&analyzer,
&[
"quick brown fox", "quick red brown", "brown quick", "quick", ],
);
assert_eq!(
search_segment(&segment, &phrase(&["quick", "brown"], 0)).expect("valid"),
vec![0]
);
assert_eq!(
search_segment(&segment, &phrase(&["quick", "brown"], 1)).expect("valid"),
vec![0, 1]
);
assert!(
!search_segment(&segment, &phrase(&["quick", "brown"], 1))
.expect("valid")
.contains(&2)
);
}
#[test]
fn a_single_term_phrase_is_just_a_term_query() {
let analyzer = code_analyzer();
let segment = segment_of(&analyzer, &["alpha beta", "gamma"]);
assert_eq!(
search_segment(&segment, &phrase(&["alpha"], 0)).expect("valid"),
vec![0]
);
}
#[test]
fn an_empty_phrase_is_a_typed_error() {
let analyzer = code_analyzer();
let segment = segment_of(&analyzer, &["alpha"]);
assert_eq!(
search_segment(&segment, &phrase(&[], 0)),
Err(PhraseError::Empty)
);
}
#[test]
fn synonym_stacked_phrase_matches_the_underscored_identifier() {
let mut vocabulary = Vocabulary::new();
vocabulary
.declare(
"put_if_match",
&[&["put", "if", "match"][..], &["putifmatch"][..]],
)
.expect("valid declaration");
let analyzer = Analyzer::new(Profile::Code.config().with_vocabulary(vocabulary))
.expect("valid config");
let segment = segment_of(&analyzer, &["call put_if_match now"]);
assert_eq!(
search_segment(&segment, &phrase(&["put", "if", "match"], 0)).expect("valid"),
vec![0]
);
assert_eq!(
search_segment(&segment, &phrase(&["put_if_match"], 0)).expect("valid"),
vec![0]
);
}
#[test]
fn a_phrase_crossing_a_stopword_gap_needs_the_slop_the_gap_costs() {
let analyzer = Analyzer::new(TokenizerConfig::text_default()).expect("valid config");
let segment = segment_of(&analyzer, &["state of the art"]);
assert!(
search_segment(&segment, &phrase(&["state", "art"], 1))
.expect("valid")
.is_empty()
);
assert_eq!(
search_segment(&segment, &phrase(&["state", "art"], 2)).expect("valid"),
vec![0]
);
}
#[test]
fn displacement_is_exact_not_greedy() {
let streams = vec![vec![0_u32, 5], vec![6]];
assert_eq!(minimum_displacement(&streams), Some(0));
assert!(streams_match(&streams, 0));
}
#[test]
fn a_missing_term_never_matches() {
assert_eq!(minimum_displacement(&[vec![1_u32], Vec::new()]), None);
assert!(!streams_match(&[vec![1_u32], Vec::new()], 100));
}
#[test]
fn stacked_positions_cost_nothing_to_cross() {
let streams = vec![vec![0_u32], vec![0], vec![1]];
assert_eq!(minimum_displacement(&streams), Some(2));
}
#[test]
fn an_empty_stream_list_yields_no_displacement() {
assert_eq!(minimum_displacement(&[]), None);
assert_eq!(minimum_displacement(&[Vec::new()]), None);
}
}