1use super::IndexedFieldMetadata;
8use super::{
9 Analyzer, Arc, BTreeMap, BlockMaxScorer, DocId, FieldName, IndexStats, PostingEntry,
10 PostingList, StorageBackendError, StorageBackendResult,
11};
12use crate::clustered_postings::BudgetedPostingReadCursor;
13use crate::clustered_postings::{
14 MaterializedPostingCursor, OccurrencePosting, PostingCursor, PostingScore,
15};
16use crate::read_control::StorageReadControl;
17use crate::TokenTermKey;
18use uqa_core::memory::Budgeted;
19use uqa_core::TokenOccurrence;
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
23#[serde(rename_all = "snake_case")]
24pub enum AnalyzerPhase {
25 Index,
27 Search,
29 Both,
31}
32
33impl AnalyzerPhase {
34 pub fn parse(s: &str) -> Result<Self, String> {
35 match s {
36 "index" => Ok(AnalyzerPhase::Index),
37 "search" | "query" => Ok(AnalyzerPhase::Search),
38 "both" => Ok(AnalyzerPhase::Both),
39 _ => Err(format!("phase must be 'index'|'search'|'both', got `{s}`")),
40 }
41 }
42}
43
44impl std::str::FromStr for AnalyzerPhase {
45 type Err = String;
46
47 fn from_str(s: &str) -> Result<Self, Self::Err> {
48 Self::parse(s)
49 }
50}
51
52pub trait InvertedIndex: Send + Sync {
53 fn source_rebuild_required(&self) -> StorageBackendResult<bool> {
55 Ok(false)
56 }
57
58 fn analyzer(&self) -> &Analyzer;
59
60 fn default_analyzer_binding(&self) -> StorageBackendResult<super::AnalyzerDefault> {
62 Err(StorageBackendError::Other(
63 "retained default analyzer bindings are not supported by this backend".into(),
64 ))
65 }
66
67 fn add_document(
68 &mut self,
69 doc_id: DocId,
70 fields: BTreeMap<FieldName, String>,
71 ) -> StorageBackendResult<()>;
72
73 fn try_add_document(
74 &mut self,
75 doc_id: DocId,
76 fields: BTreeMap<FieldName, String>,
77 ) -> StorageBackendResult<()> {
78 self.add_document(doc_id, fields)
79 }
80
81 fn try_add_documents(
83 &mut self,
84 documents: Vec<(DocId, BTreeMap<FieldName, String>)>,
85 ) -> StorageBackendResult<()> {
86 for (doc_id, fields) in documents {
87 self.try_add_document(doc_id, fields)?;
88 }
89 Ok(())
90 }
91
92 fn try_add_documents_observed(
94 &mut self,
95 _documents: Vec<(DocId, BTreeMap<FieldName, String>)>,
96 _visit: &mut super::InvertedIndexChangeVisitor<'_>,
97 ) -> StorageBackendResult<()> {
98 Err(StorageBackendError::Other(
99 "evaluated text mutation observations are not supported by this backend".into(),
100 ))
101 }
102
103 fn try_remove_document_observed(
105 &mut self,
106 doc_id: DocId,
107 visit: &mut super::InvertedIndexChangeVisitor<'_>,
108 ) -> StorageBackendResult<()> {
109 self.try_add_documents_observed(vec![(doc_id, BTreeMap::new())], visit)
110 }
111
112 fn remove_document(&mut self, doc_id: DocId) -> StorageBackendResult<()>;
113
114 fn try_remove_document(&mut self, doc_id: DocId) -> StorageBackendResult<()> {
115 self.remove_document(doc_id)
116 }
117
118 fn clear(&mut self) -> StorageBackendResult<()>;
119
120 fn try_clear(&mut self) -> StorageBackendResult<()> {
121 self.clear()
122 }
123
124 fn try_rebuild_documents(
125 &mut self,
126 documents: Vec<(DocId, BTreeMap<FieldName, String>)>,
127 ) -> StorageBackendResult<()> {
128 self.try_clear()?;
129 for (doc_id, fields) in documents {
130 if !fields.is_empty() {
131 self.try_add_document(doc_id, fields)?;
132 }
133 }
134 Ok(())
135 }
136
137 fn get_posting_list(&self, field: &str, term: &str) -> StorageBackendResult<PostingList>;
138
139 fn get_posting_list_key(
141 &self,
142 field: &str,
143 term: &TokenTermKey,
144 ) -> StorageBackendResult<PostingList> {
145 self.get_posting_list(field, &term.to_term().into_string()?)
146 }
147
148 fn posting_cursor_key(
150 &self,
151 field: &str,
152 term: &TokenTermKey,
153 ) -> StorageBackendResult<Box<dyn PostingCursor>> {
154 self.posting_cursor(field, &term.to_term().into_string()?)
155 }
156
157 fn posting_read_cursor_key<'a>(
159 &'a self,
160 field: &'a str,
161 term: &TokenTermKey,
162 ) -> StorageBackendResult<Box<dyn crate::clustered_postings::PostingReadCursor + 'a>> {
163 Ok(Box::new(crate::clustered_postings::OwnedPostingReadCursor(
164 self.posting_cursor_key(field, term)?,
165 )))
166 }
167
168 fn posting_read_cursor_key_budgeted<'a>(
170 &'a self,
171 field: &'a str,
172 term: &'a TokenTermKey,
173 control: &StorageReadControl,
174 ) -> StorageBackendResult<BudgetedPostingReadCursor<'a>> {
175 crate::clustered_postings::open_controlled_cursor(self, field, term, control)
176 }
177
178 fn visit_score_clusters(
180 &self,
181 _field: &str,
182 _term: &TokenTermKey,
183 _after: Option<u64>,
184 _limit: usize,
185 control: &StorageReadControl,
186 _visit: &mut crate::clustered_postings::ScoreClusterVisitor<'_>,
187 ) -> StorageBackendResult<()> {
188 control.check()?;
189 Err(StorageBackendError::Other(
190 "controlled score cluster reads are not supported by this backend".into(),
191 ))
192 }
193
194 fn get_occurrences_budgeted(
196 &self,
197 _doc_id: DocId,
198 _field: &str,
199 _term: &TokenTermKey,
200 control: &StorageReadControl,
201 ) -> StorageBackendResult<Budgeted<Vec<TokenOccurrence>>> {
202 control.check()?;
203 Err(StorageBackendError::Other(
204 "controlled occurrence reads are not supported by this backend".into(),
205 ))
206 }
207
208 fn get_occurrence_postings(
210 &self,
211 _field: &str,
212 _term: &TokenTermKey,
213 ) -> StorageBackendResult<Vec<OccurrencePosting>> {
214 Err(StorageBackendError::Other(
215 "lossless occurrence storage is not supported by this backend".into(),
216 ))
217 }
218
219 fn get_occurrences(
221 &self,
222 doc_id: DocId,
223 field: &str,
224 term: &TokenTermKey,
225 ) -> StorageBackendResult<Vec<TokenOccurrence>> {
226 Ok(self
227 .get_occurrence_postings(field, term)?
228 .into_iter()
229 .find(|posting| posting.doc_id == doc_id)
230 .map_or_else(Vec::new, |posting| posting.occurrences))
231 }
232
233 fn indexed_field_metadata(
235 &self,
236 _doc_id: DocId,
237 _field: &str,
238 ) -> StorageBackendResult<Option<IndexedFieldMetadata>> {
239 Err(StorageBackendError::Other(
240 "indexed field analysis metadata is not supported by this backend".into(),
241 ))
242 }
243
244 fn doc_freq_key(&self, field: &str, term: &TokenTermKey) -> StorageBackendResult<u64> {
245 self.doc_freq(field, &term.to_term().into_string()?)
246 }
247
248 fn get_term_freq_key(
249 &self,
250 doc_id: DocId,
251 field: &str,
252 term: &TokenTermKey,
253 ) -> StorageBackendResult<u64> {
254 self.get_term_freq(doc_id, field, &term.to_term().into_string()?)
255 }
256
257 fn vocabulary_keys(&self, field: &str) -> StorageBackendResult<Vec<TokenTermKey>> {
259 Ok(self
260 .vocabulary_terms(field)?
261 .iter()
262 .map(|term| TokenTermKey::from_text(term))
263 .collect())
264 }
265
266 fn get_posting_lists_bulk(
267 &self,
268 field: &str,
269 terms: &[String],
270 ) -> StorageBackendResult<Vec<PostingList>> {
271 terms
272 .iter()
273 .map(|term| self.get_posting_list(field, term))
274 .collect()
275 }
276
277 fn posting_cursor(
284 &self,
285 field: &str,
286 term: &str,
287 ) -> StorageBackendResult<Box<dyn PostingCursor>> {
288 let posting_list = self.get_posting_list(field, term)?;
289 let mut entries = Vec::with_capacity(posting_list.len());
290 for posting in posting_list {
291 let term_freq = self.get_term_freq(posting.doc_id, field, term)?;
292 entries.push(PostingScore {
293 doc_id: posting.doc_id,
294 term_freq,
295 doc_length: self.get_doc_length(posting.doc_id, field)?,
296 });
297 }
298 Ok(Box::new(MaterializedPostingCursor::new(entries)?))
299 }
300
301 fn posting_cursors_bulk(
302 &self,
303 field: &str,
304 terms: &[String],
305 ) -> StorageBackendResult<Vec<Box<dyn PostingCursor>>> {
306 terms
307 .iter()
308 .map(|term| self.posting_cursor(field, term))
309 .collect()
310 }
311
312 fn posting_cursors_keys_bulk(
314 &self,
315 field: &str,
316 terms: &[TokenTermKey],
317 ) -> StorageBackendResult<Vec<Box<dyn PostingCursor>>> {
318 if let Some(scalar) = terms
319 .iter()
320 .map(|key| key.as_str().map(str::to_owned))
321 .collect::<Option<Vec<_>>>()
322 {
323 return self.posting_cursors_bulk(field, &scalar);
324 }
325 terms
326 .iter()
327 .map(|term| self.posting_cursor_key(field, term))
328 .collect()
329 }
330
331 fn get_posting_lists_keys_bulk(
333 &self,
334 field: &str,
335 terms: &[TokenTermKey],
336 ) -> StorageBackendResult<Vec<PostingList>> {
337 if let Some(scalar) = terms
338 .iter()
339 .map(|key| key.as_str().map(str::to_owned))
340 .collect::<Option<Vec<_>>>()
341 {
342 return self.get_posting_lists_bulk(field, &scalar);
343 }
344 terms
345 .iter()
346 .map(|term| self.get_posting_list_key(field, term))
347 .collect()
348 }
349
350 fn persisted_block_max_scores_keys_bulk(
352 &self,
353 field: &str,
354 terms: &[TokenTermKey],
355 scorer_fingerprint: &str,
356 ) -> StorageBackendResult<Vec<Option<Vec<f64>>>> {
357 if let Some(scalar) = terms
358 .iter()
359 .map(|key| key.as_str().map(str::to_owned))
360 .collect::<Option<Vec<_>>>()
361 {
362 return self.persisted_block_max_scores_bulk(field, &scalar, scorer_fingerprint);
363 }
364 terms
365 .iter()
366 .map(|key| match key.as_str() {
367 Some(term) => self.persisted_block_max_scores(field, term, scorer_fingerprint),
368 None => Ok(None),
369 })
370 .collect()
371 }
372
373 fn get_scoring_inputs_keys_bulk(
375 &self,
376 doc_ids: &[DocId],
377 field: &str,
378 terms: &[TokenTermKey],
379 ) -> StorageBackendResult<Vec<(u64, Vec<u64>)>> {
380 if let Some(scalar) = terms
381 .iter()
382 .map(|key| key.as_str().map(str::to_owned))
383 .collect::<Option<Vec<_>>>()
384 {
385 return self.get_scoring_inputs_bulk(doc_ids, field, &scalar);
386 }
387 doc_ids
388 .iter()
389 .map(|id| {
390 Ok((
391 self.get_doc_length(*id, field)?,
392 terms
393 .iter()
394 .map(|key| self.get_term_freq_key(*id, field, key))
395 .collect::<StorageBackendResult<_>>()?,
396 ))
397 })
398 .collect()
399 }
400
401 fn rebuild_persisted_block_max(
407 &mut self,
408 _field: &str,
409 _scorer: &dyn BlockMaxScorer,
410 _scorer_fingerprint: &str,
411 ) -> StorageBackendResult<bool> {
412 Ok(false)
413 }
414
415 fn persisted_block_max_scores(
418 &self,
419 _field: &str,
420 _term: &str,
421 _scorer_fingerprint: &str,
422 ) -> StorageBackendResult<Option<Vec<f64>>> {
423 Ok(None)
424 }
425
426 fn persisted_block_max_scores_bulk(
428 &self,
429 field: &str,
430 terms: &[String],
431 scorer_fingerprint: &str,
432 ) -> StorageBackendResult<Vec<Option<Vec<f64>>>> {
433 terms
434 .iter()
435 .map(|term| self.persisted_block_max_scores(field, term, scorer_fingerprint))
436 .collect()
437 }
438
439 fn for_each_posting(
443 &self,
444 field: &str,
445 term: &str,
446 visit: &mut dyn FnMut(&PostingEntry),
447 ) -> StorageBackendResult<()> {
448 for entry in &self.get_posting_list(field, term)? {
449 visit(entry);
450 }
451 Ok(())
452 }
453
454 fn for_each_term_freq(
459 &self,
460 field: &str,
461 term: &str,
462 visit: &mut dyn FnMut(DocId, u64),
463 ) -> StorageBackendResult<()> {
464 for entry in &self.get_posting_list(field, term)? {
465 visit(entry.doc_id, self.get_term_freq(entry.doc_id, field, term)?);
466 }
467 Ok(())
468 }
469
470 fn doc_freq(&self, field: &str, term: &str) -> StorageBackendResult<u64>;
471
472 fn get_doc_length(&self, doc_id: DocId, field: &str) -> StorageBackendResult<u64>;
473
474 fn get_term_freq(&self, doc_id: DocId, field: &str, term: &str) -> StorageBackendResult<u64>;
475
476 fn doc_count(&self) -> StorageBackendResult<u64>;
477
478 fn total_field_length(&self, field: &str) -> StorageBackendResult<u64>;
479
480 fn field_doc_count(&self, field: &str) -> StorageBackendResult<u64> {
482 self.doc_length_count(Some(field))
483 }
484
485 fn field_stats(&self, field: &str) -> StorageBackendResult<IndexStats> {
491 super::defaults::field_stats(self, field)
492 }
493
494 fn field_stats_scalar(&self, field: &str) -> StorageBackendResult<IndexStats> {
502 super::defaults::field_stats_scalar(self, field)
503 }
504
505 fn field_stats_scalar_budgeted(
507 &self,
508 _field: &str,
509 control: &StorageReadControl,
510 ) -> StorageBackendResult<IndexStats> {
511 control.check()?;
512 Err(StorageBackendError::Other(
513 "controlled field statistics are not supported by this backend".into(),
514 ))
515 }
516
517 fn vocabulary_terms(&self, _field: &str) -> StorageBackendResult<Vec<String>> {
523 Ok(Vec::new())
524 }
525
526 fn stats(&self) -> StorageBackendResult<IndexStats>;
529
530 fn posting_count(&self, _field: Option<&str>) -> StorageBackendResult<u64> {
533 Ok(0)
534 }
535
536 fn doc_length_count(&self, _field: Option<&str>) -> StorageBackendResult<u64> {
539 Ok(0)
540 }
541
542 fn term_count(&self, _field: Option<&str>) -> StorageBackendResult<u64> {
545 Ok(0)
546 }
547
548 fn snapshot(&self) -> StorageBackendResult<Arc<dyn InvertedIndex>>;
550
551 fn snapshot_with_control(
553 &self,
554 control: &crate::read_control::StorageReadControl,
555 ) -> StorageBackendResult<Arc<dyn InvertedIndex>> {
556 control.check()?;
557 let snapshot = self.snapshot()?;
558 control.check()?;
559 Ok(snapshot)
560 }
561
562 fn writable_snapshot(&self) -> StorageBackendResult<Box<dyn InvertedIndex>> {
565 Err(StorageBackendError::Other(
566 "writable inverted-index snapshots are not supported by this backend".into(),
567 ))
568 }
569
570 fn field_names(&self) -> StorageBackendResult<Vec<FieldName>> {
576 Ok(Vec::new())
577 }
578
579 fn get_posting_list_any_field(&self, term: &str) -> StorageBackendResult<PostingList> {
583 super::defaults::get_posting_list_any_field(self, term)
584 }
585
586 fn doc_freq_any_field(&self, term: &str) -> StorageBackendResult<u64> {
588 super::defaults::doc_freq_any_field(self, term)
589 }
590
591 fn get_total_doc_length(&self, doc_id: DocId) -> StorageBackendResult<u64> {
593 super::defaults::get_total_doc_length(self, doc_id)
594 }
595
596 fn get_doc_lengths_bulk(
598 &self,
599 doc_ids: &[DocId],
600 field: &str,
601 ) -> StorageBackendResult<BTreeMap<DocId, u64>> {
602 let mut out = BTreeMap::new();
603 for doc_id in doc_ids {
604 out.insert(*doc_id, self.get_doc_length(*doc_id, field)?);
605 }
606 Ok(out)
607 }
608
609 fn get_term_freqs_bulk(
611 &self,
612 doc_ids: &[DocId],
613 field: &str,
614 term: &str,
615 ) -> StorageBackendResult<BTreeMap<DocId, u64>> {
616 super::defaults::get_term_freqs_bulk(self, doc_ids, field, term)
617 }
618
619 fn get_scoring_inputs_bulk(
624 &self,
625 doc_ids: &[DocId],
626 field: &str,
627 terms: &[String],
628 ) -> StorageBackendResult<Vec<(u64, Vec<u64>)>> {
629 let mut out = Vec::with_capacity(doc_ids.len());
630 for doc_id in doc_ids {
631 let mut term_freqs = Vec::with_capacity(terms.len());
632 for term in terms {
633 term_freqs.push(self.get_term_freq(*doc_id, field, term)?);
634 }
635 out.push((self.get_doc_length(*doc_id, field)?, term_freqs));
636 }
637 Ok(out)
638 }
639
640 fn get_total_term_freq(&self, doc_id: DocId, term: &str) -> StorageBackendResult<u64> {
643 super::defaults::get_total_term_freq(self, doc_id, term)
644 }
645
646 fn set_field_analyzer(
651 &mut self,
652 _field: &str,
653 _analyzer: Analyzer,
654 _phase: AnalyzerPhase,
655 ) -> Result<(), String> {
656 Err("set_field_analyzer not supported by this InvertedIndex backend".into())
657 }
658
659 fn remove_field_analyzers(&mut self, _field: &str) -> Result<(), String> {
664 Err("remove_field_analyzers not supported by this InvertedIndex backend".into())
665 }
666
667 fn get_field_analyzer(&self, _field: &str) -> Analyzer {
670 self.analyzer().clone()
671 }
672
673 fn get_search_analyzer(&self, field: &str) -> Analyzer {
675 self.get_field_analyzer(field)
676 }
677
678 fn index_analyzer_revision(
680 &self,
681 field: &str,
682 ) -> StorageBackendResult<Arc<uqa_analysis::CompiledAnalyzer>> {
683 Ok(self.get_field_analyzer(field).compile()?)
684 }
685
686 fn search_analyzer_revision(
688 &self,
689 field: &str,
690 ) -> StorageBackendResult<Arc<uqa_analysis::CompiledAnalyzer>> {
691 Ok(self.get_search_analyzer(field).compile()?)
692 }
693
694 fn set_field_analyzer_revision(
696 &mut self,
697 _field: &str,
698 _revision: Arc<uqa_analysis::CompiledAnalyzer>,
699 _phase: AnalyzerPhase,
700 ) -> Result<(), String> {
701 Err("immutable analyzer revisions are not supported by this backend".into())
702 }
703
704 fn set_field_analyzer_revisions(
706 &mut self,
707 _field: &str,
708 _index: Arc<uqa_analysis::CompiledAnalyzer>,
709 _search: Arc<uqa_analysis::CompiledAnalyzer>,
710 ) -> Result<(), String> {
711 Err("atomic analyzer revision pairs are not supported by this backend".into())
712 }
713
714 fn rebuild_with_analyzer_revision(
716 &mut self,
717 _field: &str,
718 _revision: Arc<uqa_analysis::CompiledAnalyzer>,
719 _phase: AnalyzerPhase,
720 _documents: Vec<(DocId, BTreeMap<FieldName, String>)>,
721 ) -> StorageBackendResult<()> {
722 Err(StorageBackendError::Other(
723 "atomic analyzer revision rebuild is not supported by this backend".into(),
724 ))
725 }
726
727 fn try_rebuild_documents_cancellable(
729 &mut self,
730 _documents: Vec<(DocId, BTreeMap<FieldName, String>)>,
731 cancellation: &uqa_core::CancellationToken,
732 ) -> StorageBackendResult<()> {
733 cancellation.check()?;
734 Err(StorageBackendError::Other(
735 "cancellable atomic index rebuild is not supported by this backend".into(),
736 ))
737 }
738
739 fn rebuild_with_analyzer_revision_cancellable(
741 &mut self,
742 _field: &str,
743 _revision: Arc<uqa_analysis::CompiledAnalyzer>,
744 _phase: AnalyzerPhase,
745 _documents: Vec<(DocId, BTreeMap<FieldName, String>)>,
746 cancellation: &uqa_core::CancellationToken,
747 ) -> StorageBackendResult<()> {
748 cancellation.check()?;
749 Err(StorageBackendError::Other(
750 "cancellable atomic analyzer revision rebuild is not supported by this backend".into(),
751 ))
752 }
753}