1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
//! Phrase query - matches documents containing terms in consecutive positions
use std::sync::Arc;
use crate::dsl::Field;
use crate::segment::SegmentReader;
use crate::structures::{BlockPostingIterator, BlockPostingList, PositionPostingList, TERMINATED};
use crate::{DocId, Score};
use super::{CountFuture, EmptyScorer, GlobalStats, Query, Scorer, ScorerFuture};
/// Phrase query - matches documents containing terms in consecutive positions
///
/// Example: "quick brown fox" matches only if all three terms appear
/// consecutively in the document.
#[derive(Clone)]
pub struct PhraseQuery {
pub field: Field,
/// Terms in the phrase, in order
pub terms: Vec<Vec<u8>>,
/// Optional slop (max distance between terms, 0 = exact phrase)
pub slop: u32,
/// Optional global statistics for cross-segment IDF
global_stats: Option<Arc<GlobalStats>>,
}
impl std::fmt::Debug for PhraseQuery {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let terms: Vec<_> = self
.terms
.iter()
.map(|t| String::from_utf8_lossy(t).to_string())
.collect();
f.debug_struct("PhraseQuery")
.field("field", &self.field)
.field("terms", &terms)
.field("slop", &self.slop)
.finish()
}
}
impl PhraseQuery {
/// Create a new exact phrase query
pub fn new(field: Field, terms: Vec<Vec<u8>>) -> Self {
Self {
field,
terms,
slop: 0,
global_stats: None,
}
}
/// Create from text (splits on whitespace and lowercases)
pub fn text(field: Field, phrase: &str) -> Self {
let terms: Vec<Vec<u8>> = phrase
.split_whitespace()
.map(|w| w.to_lowercase().into_bytes())
.collect();
Self {
field,
terms,
slop: 0,
global_stats: None,
}
}
/// Set slop (max distance between terms)
pub fn with_slop(mut self, slop: u32) -> Self {
self.slop = slop;
self
}
/// Set global statistics for cross-segment IDF
pub fn with_global_stats(mut self, stats: Arc<GlobalStats>) -> Self {
self.global_stats = Some(stats);
self
}
}
impl Query for PhraseQuery {
fn scorer<'a>(&self, reader: &'a SegmentReader, limit: usize) -> ScorerFuture<'a> {
let field = self.field;
let terms = self.terms.clone();
let slop = self.slop;
let _global_stats = self.global_stats.clone();
Box::pin(async move {
if terms.is_empty() {
return Ok(Box::new(EmptyScorer) as Box<dyn Scorer + 'a>);
}
// Single term - delegate to TermQuery
if terms.len() == 1 {
let term_query = super::TermQuery::new(field, terms[0].clone());
return term_query.scorer(reader, limit).await;
}
// Check if positions are available
if !reader.has_positions(field) {
// Fall back to AND query (BooleanQuery with MUST clauses)
let mut bool_query = super::BooleanQuery::new();
for term in &terms {
bool_query = bool_query.must(super::TermQuery::new(field, term.clone()));
}
return bool_query.scorer(reader, limit).await;
}
// Load postings and positions for all terms (parallel per term)
let mut term_postings: Vec<BlockPostingList> = Vec::with_capacity(terms.len());
let mut term_positions: Vec<PositionPostingList> = Vec::with_capacity(terms.len());
for term in &terms {
// Fetch postings and positions in parallel
let (postings, positions) = futures::join!(
reader.get_postings(field, term),
reader.get_positions(field, term)
);
match (postings?, positions?) {
(Some(p), Some(pos)) => {
term_postings.push(p);
term_positions.push(pos);
}
_ => {
// If any term is missing, no documents can match
return Ok(Box::new(EmptyScorer) as Box<dyn Scorer + 'a>);
}
}
}
// Compute combined IDF (sum of individual IDFs)
let idf: f32 = term_postings
.iter()
.map(|p| {
let num_docs = reader.num_docs() as f32;
let doc_freq = p.doc_count() as f32;
super::bm25_idf(doc_freq, num_docs)
})
.sum();
let avg_field_len = reader.avg_field_len(field);
Ok(Box::new(PhraseScorer::new(
term_postings,
term_positions,
slop,
idf,
avg_field_len,
)) as Box<dyn Scorer + 'a>)
})
}
fn count_estimate<'a>(&self, reader: &'a SegmentReader) -> CountFuture<'a> {
let field = self.field;
let terms = self.terms.clone();
Box::pin(async move {
if terms.is_empty() {
return Ok(0);
}
// Estimate based on minimum posting list size
let mut min_count = u32::MAX;
for term in &terms {
match reader.get_postings(field, term).await? {
Some(list) => min_count = min_count.min(list.doc_count()),
None => return Ok(0),
}
}
// Phrase matching will typically match fewer docs than the minimum
// Estimate ~10% of the smallest posting list
Ok((min_count / 10).max(1))
})
}
}
/// Scorer that checks phrase positions
struct PhraseScorer {
/// Posting iterators for each term
posting_iters: Vec<BlockPostingIterator<'static>>,
/// Position iterators for each term
position_lists: Vec<PositionPostingList>,
/// Max slop between terms
slop: u32,
/// Current matching document
current_doc: DocId,
/// Combined IDF
idf: f32,
/// Average field length
avg_field_len: f32,
}
impl PhraseScorer {
fn new(
posting_lists: Vec<BlockPostingList>,
position_lists: Vec<PositionPostingList>,
slop: u32,
idf: f32,
avg_field_len: f32,
) -> Self {
let posting_iters: Vec<_> = posting_lists
.into_iter()
.map(|p| p.into_iterator())
.collect();
let mut scorer = Self {
posting_iters,
position_lists,
slop,
current_doc: 0,
idf,
avg_field_len,
};
scorer.find_next_phrase_match();
scorer
}
/// Find next document where all terms appear as a phrase
fn find_next_phrase_match(&mut self) {
loop {
// First, find a document where all terms appear (AND semantics)
let doc = self.find_next_and_match();
if doc == TERMINATED {
self.current_doc = TERMINATED;
return;
}
// Check if positions form a valid phrase
if self.check_phrase_positions(doc) {
self.current_doc = doc;
return;
}
// Advance and try again
self.posting_iters[0].advance();
}
}
/// Find next document where all terms appear
fn find_next_and_match(&mut self) -> DocId {
if self.posting_iters.is_empty() {
return TERMINATED;
}
loop {
let max_doc = self.posting_iters.iter().map(|it| it.doc()).max().unwrap();
if max_doc == TERMINATED {
return TERMINATED;
}
let mut all_match = true;
for it in &mut self.posting_iters {
let doc = it.seek(max_doc);
if doc != max_doc {
all_match = false;
if doc == TERMINATED {
return TERMINATED;
}
}
}
if all_match {
return max_doc;
}
}
}
/// Check if positions form a valid phrase for the given document
fn check_phrase_positions(&self, doc_id: DocId) -> bool {
// Get positions for each term in this document
let mut term_positions: Vec<Vec<u32>> = Vec::with_capacity(self.position_lists.len());
for pos_list in &self.position_lists {
match pos_list.get_positions(doc_id) {
Some(positions) => term_positions.push(positions.to_vec()),
None => return false,
}
}
// Check for consecutive positions
// For exact phrase (slop=0), position[i+1] = position[i] + 1
self.find_phrase_match(&term_positions)
}
/// Find if there's a valid phrase match among the positions
fn find_phrase_match(&self, term_positions: &[Vec<u32>]) -> bool {
if term_positions.is_empty() {
return false;
}
// For each position of the first term, check if subsequent terms
// have positions that form a phrase
for &first_pos in &term_positions[0] {
if self.check_phrase_from_position(first_pos, term_positions) {
return true;
}
}
false
}
/// Check if a phrase exists starting from the given position
fn check_phrase_from_position(&self, start_pos: u32, term_positions: &[Vec<u32>]) -> bool {
let mut expected_pos = start_pos;
for (i, positions) in term_positions.iter().enumerate() {
if i == 0 {
continue; // Skip first term, already matched
}
expected_pos += 1;
// Find a position within slop distance
let found = positions.iter().any(|&pos| {
if self.slop == 0 {
pos == expected_pos
} else {
let diff = pos.abs_diff(expected_pos);
diff <= self.slop
}
});
if !found {
return false;
}
}
true
}
}
impl Scorer for PhraseScorer {
fn doc(&self) -> DocId {
self.current_doc
}
fn score(&self) -> Score {
if self.current_doc == TERMINATED {
return 0.0;
}
// Sum term frequencies for BM25 scoring
let tf: f32 = self
.posting_iters
.iter()
.map(|it| it.term_freq() as f32)
.sum();
// Phrase matches get a boost since they're more precise
super::bm25_score(tf, self.idf, tf, self.avg_field_len) * 1.5
}
fn advance(&mut self) -> DocId {
if self.current_doc == TERMINATED {
return TERMINATED;
}
self.posting_iters[0].advance();
self.find_next_phrase_match();
self.current_doc
}
fn seek(&mut self, target: DocId) -> DocId {
if target == TERMINATED {
self.current_doc = TERMINATED;
return TERMINATED;
}
self.posting_iters[0].seek(target);
self.find_next_phrase_match();
self.current_doc
}
fn size_hint(&self) -> u32 {
0
}
}