1use super::{
8 Analyzer, Arc, BTreeMap, BlockMaxScorer, DocId, FieldName, IndexStats, PostingEntry,
9 PostingList, StorageBackendError, StorageBackendResult,
10};
11use super::{IndexedFieldMetadata, TextIndexSource};
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(
126 &mut self,
127 source: &mut dyn TextIndexSource,
128 ) -> StorageBackendResult<()> {
129 self.try_clear()?;
130 while let Some((doc_id, fields)) = source.next_document()? {
131 if !fields.is_empty() {
132 self.try_add_document(doc_id, fields)?;
133 }
134 }
135 Ok(())
136 }
137
138 fn get_posting_list(&self, field: &str, term: &str) -> StorageBackendResult<PostingList>;
139
140 fn get_posting_list_key(
142 &self,
143 field: &str,
144 term: &TokenTermKey,
145 ) -> StorageBackendResult<PostingList> {
146 self.get_posting_list(field, &term.to_term().into_string()?)
147 }
148
149 fn posting_cursor_key(
151 &self,
152 field: &str,
153 term: &TokenTermKey,
154 ) -> StorageBackendResult<Box<dyn PostingCursor>> {
155 self.posting_cursor(field, &term.to_term().into_string()?)
156 }
157
158 fn posting_read_cursor_key<'a>(
160 &'a self,
161 field: &'a str,
162 term: &TokenTermKey,
163 ) -> StorageBackendResult<Box<dyn crate::clustered_postings::PostingReadCursor + 'a>> {
164 Ok(Box::new(crate::clustered_postings::OwnedPostingReadCursor(
165 self.posting_cursor_key(field, term)?,
166 )))
167 }
168
169 fn posting_read_cursor_key_budgeted<'a>(
171 &'a self,
172 field: &'a str,
173 term: &'a TokenTermKey,
174 control: &StorageReadControl,
175 ) -> StorageBackendResult<BudgetedPostingReadCursor<'a>> {
176 crate::clustered_postings::open_controlled_cursor(self, field, term, control)
177 }
178
179 fn visit_score_clusters(
181 &self,
182 _field: &str,
183 _term: &TokenTermKey,
184 _after: Option<u64>,
185 _limit: usize,
186 control: &StorageReadControl,
187 _visit: &mut crate::clustered_postings::ScoreClusterVisitor<'_>,
188 ) -> StorageBackendResult<()> {
189 control.check()?;
190 Err(StorageBackendError::Other(
191 "controlled score cluster reads are not supported by this backend".into(),
192 ))
193 }
194
195 fn get_occurrences_budgeted(
197 &self,
198 _doc_id: DocId,
199 _field: &str,
200 _term: &TokenTermKey,
201 control: &StorageReadControl,
202 ) -> StorageBackendResult<Budgeted<Vec<TokenOccurrence>>> {
203 control.check()?;
204 Err(StorageBackendError::Other(
205 "controlled occurrence reads are not supported by this backend".into(),
206 ))
207 }
208
209 fn get_occurrence_postings(
211 &self,
212 _field: &str,
213 _term: &TokenTermKey,
214 ) -> StorageBackendResult<Vec<OccurrencePosting>> {
215 Err(StorageBackendError::Other(
216 "lossless occurrence storage is not supported by this backend".into(),
217 ))
218 }
219
220 fn get_occurrences(
222 &self,
223 doc_id: DocId,
224 field: &str,
225 term: &TokenTermKey,
226 ) -> StorageBackendResult<Vec<TokenOccurrence>> {
227 Ok(self
228 .get_occurrence_postings(field, term)?
229 .into_iter()
230 .find(|posting| posting.doc_id == doc_id)
231 .map_or_else(Vec::new, |posting| posting.occurrences))
232 }
233
234 fn indexed_field_metadata(
236 &self,
237 _doc_id: DocId,
238 _field: &str,
239 ) -> StorageBackendResult<Option<IndexedFieldMetadata>> {
240 Err(StorageBackendError::Other(
241 "indexed field analysis metadata is not supported by this backend".into(),
242 ))
243 }
244
245 fn doc_freq_key(&self, field: &str, term: &TokenTermKey) -> StorageBackendResult<u64> {
246 self.doc_freq(field, &term.to_term().into_string()?)
247 }
248
249 fn get_term_freq_key(
250 &self,
251 doc_id: DocId,
252 field: &str,
253 term: &TokenTermKey,
254 ) -> StorageBackendResult<u64> {
255 self.get_term_freq(doc_id, field, &term.to_term().into_string()?)
256 }
257
258 fn vocabulary_keys(&self, field: &str) -> StorageBackendResult<Vec<TokenTermKey>> {
260 Ok(self
261 .vocabulary_terms(field)?
262 .iter()
263 .map(|term| TokenTermKey::from_text(term))
264 .collect())
265 }
266
267 fn get_posting_lists_bulk(
268 &self,
269 field: &str,
270 terms: &[String],
271 ) -> StorageBackendResult<Vec<PostingList>> {
272 terms
273 .iter()
274 .map(|term| self.get_posting_list(field, term))
275 .collect()
276 }
277
278 fn posting_cursor(
285 &self,
286 field: &str,
287 term: &str,
288 ) -> StorageBackendResult<Box<dyn PostingCursor>> {
289 let posting_list = self.get_posting_list(field, term)?;
290 let mut entries = Vec::with_capacity(posting_list.len());
291 for posting in posting_list {
292 let term_freq = self.get_term_freq(posting.doc_id, field, term)?;
293 entries.push(PostingScore {
294 doc_id: posting.doc_id,
295 term_freq,
296 doc_length: self.get_doc_length(posting.doc_id, field)?,
297 });
298 }
299 Ok(Box::new(MaterializedPostingCursor::new(entries)?))
300 }
301
302 fn posting_cursors_bulk(
303 &self,
304 field: &str,
305 terms: &[String],
306 ) -> StorageBackendResult<Vec<Box<dyn PostingCursor>>> {
307 terms
308 .iter()
309 .map(|term| self.posting_cursor(field, term))
310 .collect()
311 }
312
313 fn posting_cursors_keys_bulk(
315 &self,
316 field: &str,
317 terms: &[TokenTermKey],
318 ) -> StorageBackendResult<Vec<Box<dyn PostingCursor>>> {
319 if let Some(scalar) = terms
320 .iter()
321 .map(|key| key.as_str().map(str::to_owned))
322 .collect::<Option<Vec<_>>>()
323 {
324 return self.posting_cursors_bulk(field, &scalar);
325 }
326 terms
327 .iter()
328 .map(|term| self.posting_cursor_key(field, term))
329 .collect()
330 }
331
332 fn get_posting_lists_keys_bulk(
334 &self,
335 field: &str,
336 terms: &[TokenTermKey],
337 ) -> StorageBackendResult<Vec<PostingList>> {
338 if let Some(scalar) = terms
339 .iter()
340 .map(|key| key.as_str().map(str::to_owned))
341 .collect::<Option<Vec<_>>>()
342 {
343 return self.get_posting_lists_bulk(field, &scalar);
344 }
345 terms
346 .iter()
347 .map(|term| self.get_posting_list_key(field, term))
348 .collect()
349 }
350
351 fn persisted_block_max_scores_keys_bulk(
353 &self,
354 field: &str,
355 terms: &[TokenTermKey],
356 scorer_fingerprint: &str,
357 ) -> StorageBackendResult<Vec<Option<Vec<f64>>>> {
358 if let Some(scalar) = terms
359 .iter()
360 .map(|key| key.as_str().map(str::to_owned))
361 .collect::<Option<Vec<_>>>()
362 {
363 return self.persisted_block_max_scores_bulk(field, &scalar, scorer_fingerprint);
364 }
365 terms
366 .iter()
367 .map(|key| match key.as_str() {
368 Some(term) => self.persisted_block_max_scores(field, term, scorer_fingerprint),
369 None => Ok(None),
370 })
371 .collect()
372 }
373
374 fn get_scoring_inputs_keys_bulk(
376 &self,
377 doc_ids: &[DocId],
378 field: &str,
379 terms: &[TokenTermKey],
380 ) -> StorageBackendResult<Vec<(u64, Vec<u64>)>> {
381 if let Some(scalar) = terms
382 .iter()
383 .map(|key| key.as_str().map(str::to_owned))
384 .collect::<Option<Vec<_>>>()
385 {
386 return self.get_scoring_inputs_bulk(doc_ids, field, &scalar);
387 }
388 doc_ids
389 .iter()
390 .map(|id| {
391 Ok((
392 self.get_doc_length(*id, field)?,
393 terms
394 .iter()
395 .map(|key| self.get_term_freq_key(*id, field, key))
396 .collect::<StorageBackendResult<_>>()?,
397 ))
398 })
399 .collect()
400 }
401
402 fn rebuild_persisted_block_max(
408 &mut self,
409 _field: &str,
410 _scorer: &dyn BlockMaxScorer,
411 _scorer_fingerprint: &str,
412 ) -> StorageBackendResult<bool> {
413 Ok(false)
414 }
415
416 fn persisted_block_max_scores(
419 &self,
420 _field: &str,
421 _term: &str,
422 _scorer_fingerprint: &str,
423 ) -> StorageBackendResult<Option<Vec<f64>>> {
424 Ok(None)
425 }
426
427 fn persisted_block_max_scores_bulk(
429 &self,
430 field: &str,
431 terms: &[String],
432 scorer_fingerprint: &str,
433 ) -> StorageBackendResult<Vec<Option<Vec<f64>>>> {
434 terms
435 .iter()
436 .map(|term| self.persisted_block_max_scores(field, term, scorer_fingerprint))
437 .collect()
438 }
439
440 fn for_each_posting(
444 &self,
445 field: &str,
446 term: &str,
447 visit: &mut dyn FnMut(&PostingEntry),
448 ) -> StorageBackendResult<()> {
449 for entry in &self.get_posting_list(field, term)? {
450 visit(entry);
451 }
452 Ok(())
453 }
454
455 fn for_each_term_freq(
460 &self,
461 field: &str,
462 term: &str,
463 visit: &mut dyn FnMut(DocId, u64),
464 ) -> StorageBackendResult<()> {
465 for entry in &self.get_posting_list(field, term)? {
466 visit(entry.doc_id, self.get_term_freq(entry.doc_id, field, term)?);
467 }
468 Ok(())
469 }
470
471 fn doc_freq(&self, field: &str, term: &str) -> StorageBackendResult<u64>;
472
473 fn get_doc_length(&self, doc_id: DocId, field: &str) -> StorageBackendResult<u64>;
474
475 fn get_term_freq(&self, doc_id: DocId, field: &str, term: &str) -> StorageBackendResult<u64>;
476
477 fn doc_count(&self) -> StorageBackendResult<u64>;
478
479 fn total_field_length(&self, field: &str) -> StorageBackendResult<u64>;
480
481 fn field_doc_count(&self, field: &str) -> StorageBackendResult<u64> {
483 self.doc_length_count(Some(field))
484 }
485
486 fn field_stats(&self, field: &str) -> StorageBackendResult<IndexStats> {
492 super::defaults::field_stats(self, field)
493 }
494
495 fn field_stats_scalar(&self, field: &str) -> StorageBackendResult<IndexStats> {
503 super::defaults::field_stats_scalar(self, field)
504 }
505
506 fn field_stats_scalar_budgeted(
508 &self,
509 _field: &str,
510 control: &StorageReadControl,
511 ) -> StorageBackendResult<IndexStats> {
512 control.check()?;
513 Err(StorageBackendError::Other(
514 "controlled field statistics are not supported by this backend".into(),
515 ))
516 }
517
518 fn vocabulary_terms(&self, _field: &str) -> StorageBackendResult<Vec<String>> {
524 Ok(Vec::new())
525 }
526
527 fn stats(&self) -> StorageBackendResult<IndexStats>;
530
531 fn posting_count(&self, _field: Option<&str>) -> StorageBackendResult<u64> {
534 Ok(0)
535 }
536
537 fn doc_length_count(&self, _field: Option<&str>) -> StorageBackendResult<u64> {
540 Ok(0)
541 }
542
543 fn term_count(&self, _field: Option<&str>) -> StorageBackendResult<u64> {
546 Ok(0)
547 }
548
549 fn snapshot(&self) -> StorageBackendResult<Arc<dyn InvertedIndex>>;
551
552 fn snapshot_with_control(
554 &self,
555 control: &crate::read_control::StorageReadControl,
556 ) -> StorageBackendResult<Arc<dyn InvertedIndex>> {
557 control.check()?;
558 let snapshot = self.snapshot()?;
559 control.check()?;
560 Ok(snapshot)
561 }
562
563 fn writable_snapshot(&self) -> StorageBackendResult<Box<dyn InvertedIndex>> {
566 Err(StorageBackendError::Other(
567 "writable inverted-index snapshots are not supported by this backend".into(),
568 ))
569 }
570
571 fn field_names(&self) -> StorageBackendResult<Vec<FieldName>> {
577 Ok(Vec::new())
578 }
579
580 fn get_posting_list_any_field(&self, term: &str) -> StorageBackendResult<PostingList> {
584 super::defaults::get_posting_list_any_field(self, term)
585 }
586
587 fn doc_freq_any_field(&self, term: &str) -> StorageBackendResult<u64> {
589 super::defaults::doc_freq_any_field(self, term)
590 }
591
592 fn get_total_doc_length(&self, doc_id: DocId) -> StorageBackendResult<u64> {
594 super::defaults::get_total_doc_length(self, doc_id)
595 }
596
597 fn get_doc_lengths_bulk(
599 &self,
600 doc_ids: &[DocId],
601 field: &str,
602 ) -> StorageBackendResult<BTreeMap<DocId, u64>> {
603 let mut out = BTreeMap::new();
604 for doc_id in doc_ids {
605 out.insert(*doc_id, self.get_doc_length(*doc_id, field)?);
606 }
607 Ok(out)
608 }
609
610 fn get_term_freqs_bulk(
612 &self,
613 doc_ids: &[DocId],
614 field: &str,
615 term: &str,
616 ) -> StorageBackendResult<BTreeMap<DocId, u64>> {
617 super::defaults::get_term_freqs_bulk(self, doc_ids, field, term)
618 }
619
620 fn get_scoring_inputs_bulk(
625 &self,
626 doc_ids: &[DocId],
627 field: &str,
628 terms: &[String],
629 ) -> StorageBackendResult<Vec<(u64, Vec<u64>)>> {
630 let mut out = Vec::with_capacity(doc_ids.len());
631 for doc_id in doc_ids {
632 let mut term_freqs = Vec::with_capacity(terms.len());
633 for term in terms {
634 term_freqs.push(self.get_term_freq(*doc_id, field, term)?);
635 }
636 out.push((self.get_doc_length(*doc_id, field)?, term_freqs));
637 }
638 Ok(out)
639 }
640
641 fn get_total_term_freq(&self, doc_id: DocId, term: &str) -> StorageBackendResult<u64> {
644 super::defaults::get_total_term_freq(self, doc_id, term)
645 }
646
647 fn set_field_analyzer(
652 &mut self,
653 _field: &str,
654 _analyzer: Analyzer,
655 _phase: AnalyzerPhase,
656 ) -> Result<(), String> {
657 Err("set_field_analyzer not supported by this InvertedIndex backend".into())
658 }
659
660 fn remove_field_analyzers(&mut self, _field: &str) -> Result<(), String> {
665 Err("remove_field_analyzers not supported by this InvertedIndex backend".into())
666 }
667
668 fn get_field_analyzer(&self, _field: &str) -> Analyzer {
671 self.analyzer().clone()
672 }
673
674 fn get_search_analyzer(&self, field: &str) -> Analyzer {
676 self.get_field_analyzer(field)
677 }
678
679 fn index_analyzer_revision(
681 &self,
682 field: &str,
683 ) -> StorageBackendResult<Arc<uqa_analysis::CompiledAnalyzer>> {
684 Ok(self.get_field_analyzer(field).compile()?)
685 }
686
687 fn search_analyzer_revision(
689 &self,
690 field: &str,
691 ) -> StorageBackendResult<Arc<uqa_analysis::CompiledAnalyzer>> {
692 Ok(self.get_search_analyzer(field).compile()?)
693 }
694
695 fn set_field_analyzer_revision(
697 &mut self,
698 _field: &str,
699 _revision: Arc<uqa_analysis::CompiledAnalyzer>,
700 _phase: AnalyzerPhase,
701 ) -> Result<(), String> {
702 Err("immutable analyzer revisions are not supported by this backend".into())
703 }
704
705 fn set_field_analyzer_revisions(
707 &mut self,
708 _field: &str,
709 _index: Arc<uqa_analysis::CompiledAnalyzer>,
710 _search: Arc<uqa_analysis::CompiledAnalyzer>,
711 ) -> Result<(), String> {
712 Err("atomic analyzer revision pairs are not supported by this backend".into())
713 }
714
715 fn rebuild_with_analyzer_revision(
717 &mut self,
718 _field: &str,
719 _revision: Arc<uqa_analysis::CompiledAnalyzer>,
720 _phase: AnalyzerPhase,
721 _source: &mut dyn TextIndexSource,
722 ) -> StorageBackendResult<()> {
723 Err(StorageBackendError::Other(
724 "atomic analyzer revision rebuild is not supported by this backend".into(),
725 ))
726 }
727
728 fn try_rebuild_documents_cancellable(
730 &mut self,
731 _source: &mut dyn TextIndexSource,
732 cancellation: &uqa_core::CancellationToken,
733 ) -> StorageBackendResult<()> {
734 cancellation.check()?;
735 Err(StorageBackendError::Other(
736 "cancellable atomic index rebuild is not supported by this backend".into(),
737 ))
738 }
739
740 fn rebuild_with_analyzer_revision_cancellable(
742 &mut self,
743 _field: &str,
744 _revision: Arc<uqa_analysis::CompiledAnalyzer>,
745 _phase: AnalyzerPhase,
746 _source: &mut dyn TextIndexSource,
747 cancellation: &uqa_core::CancellationToken,
748 ) -> StorageBackendResult<()> {
749 cancellation.check()?;
750 Err(StorageBackendError::Other(
751 "cancellable atomic analyzer revision rebuild is not supported by this backend".into(),
752 ))
753 }
754}