zeppelin_embed/lifecycle/
materialize.rs

1//! Scoped ownership of text returned from a ranked store query.
2
3use super::{PublishedSnapshot, QueryCancellation, QueryError, SnapshotLease, Store, StoreError};
4use crate::ingest::{
5    ActiveSegment, DocId, DocumentVersion, GlobalRowId, RowSource, SearchCandidate,
6};
7use std::cell::Cell;
8use std::num::NonZeroUsize;
9use std::sync::Arc;
10
11/// Per-hit hybrid snippets in rank order, or the failure to read a ranked
12/// row's text. `None` marks a hit without stored text or without a match.
13pub type HybridSnippets =
14    Result<Vec<Option<crate::fts::query::OwnedLexicalSnippet>>, MaterializationError>;
15
16/// Preserves a deferred vector producer's typed error separately from search.
17#[derive(Debug)]
18pub enum HybridPreparationError<E> {
19    /// The caller could not produce a validated query vector.
20    Preparation(E),
21    /// Admission, retrieval, fusion, cancellation or materialization failed.
22    Search(crate::fusion::FusionError),
23}
24
25impl<E> From<crate::fusion::FusionError> for HybridPreparationError<E> {
26    fn from(error: crate::fusion::FusionError) -> Self {
27        Self::Search(error)
28    }
29}
30
31impl<E> From<QueryError> for HybridPreparationError<E> {
32    fn from(error: QueryError) -> Self {
33        Self::Search(error.into())
34    }
35}
36
37impl<E: std::fmt::Display> std::fmt::Display for HybridPreparationError<E> {
38    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
39        match self {
40            Self::Preparation(error) => error.fmt(formatter),
41            Self::Search(error) => error.fmt(formatter),
42        }
43    }
44}
45
46impl<E: std::error::Error + 'static> std::error::Error for HybridPreparationError<E> {
47    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
48        match self {
49            Self::Preparation(error) => Some(error),
50            Self::Search(error) => Some(error),
51        }
52    }
53}
54
55#[cfg(any(test, feature = "test-seams"))]
56#[derive(Default)]
57pub(super) struct TestMaterializationWork {
58    pub(super) admissions: std::sync::atomic::AtomicU64,
59    pub(super) reverse_version_lookups: std::sync::atomic::AtomicU64,
60    pub(super) direct_row_lookups: std::sync::atomic::AtomicU64,
61    #[cfg(feature = "query-timing")]
62    pub(super) lexical_admission_locks: std::sync::atomic::AtomicU64,
63    #[cfg(feature = "query-timing")]
64    pub(super) lexical_admission_lock_nanos: std::sync::atomic::AtomicU64,
65}
66
67#[cfg(all(feature = "query-timing", any(test, feature = "test-seams")))]
68impl TestMaterializationWork {
69    pub(super) fn observe_lexical_lock(&self, started: std::time::Instant) {
70        use std::sync::atomic::Ordering::Relaxed;
71        self.lexical_admission_locks.fetch_add(1, Relaxed);
72        self.lexical_admission_lock_nanos.fetch_add(
73            started.elapsed().as_nanos().min(u128::from(u64::MAX)) as u64,
74            Relaxed,
75        );
76    }
77}
78
79/// Literal lifecycle and source-lookup observations for directed query tests.
80#[cfg(any(test, feature = "test-seams"))]
81#[derive(Clone, Copy, Debug, Eq, PartialEq)]
82pub struct MaterializationTestCounters {
83    /// Invocations of vector or lexical admission, including standalone stored_text.
84    pub admissions: u64,
85    /// Invocations of active or sealed reverse document-version lookup.
86    pub reverse_version_lookups: u64,
87    /// Ranked physical rows requested through the scoped materializer.
88    pub direct_row_lookups: u64,
89    /// Measured state, active, and snapshot lock acquisitions in lexical admission.
90    #[cfg(feature = "query-timing")]
91    pub lexical_admission_locks: u64,
92    /// Wall nanoseconds in those acquisitions, including uncontended call overhead.
93    #[cfg(feature = "query-timing")]
94    pub lexical_admission_lock_nanos: u64,
95}
96
97/// One owned source row corresponding to a ranked query hit.
98#[derive(Clone, Debug, Eq, PartialEq)]
99pub struct MaterializedRow {
100    /// The exact document version whose row was scored.
101    pub document: DocumentVersion,
102    /// Original physical row coordinates within the retrieval snapshot.
103    pub row_id: GlobalRowId,
104    /// One owned copy of the stored UTF-8 body.
105    pub text: String,
106}
107
108/// A ranked row could not be materialized faithfully.
109#[derive(Debug)]
110pub enum MaterializationError {
111    /// The caller requested a rank outside the returned list.
112    RankOutOfRange {
113        /// Zero-based rank requested by the caller.
114        rank: usize,
115        /// Number of ranked hits available.
116        returned: usize,
117    },
118    /// The ranked row has no application identity.
119    MissingIdentity {
120        /// Physical row lacking an application identity.
121        row_id: GlobalRowId,
122    },
123    /// The ranked document has no stored text.
124    MissingText {
125        /// Physical row lacking stored text.
126        row_id: GlobalRowId,
127        /// Ranked application identity.
128        document: DocumentVersion,
129    },
130    /// The ranked physical source is absent from the admitted snapshot.
131    MissingSource {
132        /// Physical row whose source is absent.
133        row_id: GlobalRowId,
134    },
135    /// A lexical source ordinal has no physical source in the admitted assembly.
136    InvalidLexicalSource {
137        /// Source ordinal from the ranked lexical hit.
138        source: u32,
139    },
140    /// A fused application key has no retained source identity.
141    MissingFusedIdentity {
142        /// Fused application key lacking source coordinates.
143        document: DocId,
144    },
145    /// The physical row does not contain the version that was ranked.
146    IdentityMismatch {
147        /// Physical row that failed identity validation.
148        row_id: GlobalRowId,
149        /// Application identity retained during ranking.
150        expected: DocumentVersion,
151        /// Application identity found at the physical row.
152        actual: Option<DocumentVersion>,
153    },
154    /// The result could not reserve its owned text buffer.
155    AllocationFailed {
156        /// UTF-8 text bytes requested from the allocator.
157        bytes: usize,
158    },
159    /// A materialization work count overflowed.
160    ArithmeticOverflow,
161    /// Cancellation or deadline prevented complete materialization.
162    Query(QueryError),
163    /// Storage validation or lifecycle failure.
164    Storage(StoreError),
165}
166
167#[derive(Clone, Copy, Debug, Eq, PartialEq)]
168pub(crate) struct VersionMismatch {
169    pub(crate) expected: DocumentVersion,
170    pub(crate) actual: Option<DocumentVersion>,
171}
172
173pub(crate) fn require_document_version(
174    expected: DocumentVersion,
175    actual: Option<DocumentVersion>,
176) -> Result<(), VersionMismatch> {
177    if actual == Some(expected) {
178        Ok(())
179    } else {
180        Err(VersionMismatch { expected, actual })
181    }
182}
183
184impl std::fmt::Display for MaterializationError {
185    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
186        match self {
187            Self::RankOutOfRange { rank, returned } => {
188                write!(f, "result rank {rank} is outside {returned} returned rows")
189            }
190            Self::MissingIdentity { row_id } => {
191                write!(f, "result row {row_id:?} has no document identity")
192            }
193            Self::MissingText { row_id, document } => write!(
194                f,
195                "result row {row_id:?} for {document:?} has no stored text"
196            ),
197            Self::MissingSource { row_id } => {
198                write!(f, "result row {row_id:?} is outside the admitted snapshot")
199            }
200            Self::InvalidLexicalSource { source } => write!(
201                f,
202                "lexical result source {source} is outside the admitted assembly"
203            ),
204            Self::MissingFusedIdentity { document } => write!(
205                f,
206                "fused document {document:?} has no ranked source identity"
207            ),
208            Self::IdentityMismatch {
209                row_id,
210                expected,
211                actual,
212            } => write!(
213                f,
214                "result row {row_id:?} expected {expected:?}, found {actual:?}"
215            ),
216            Self::AllocationFailed { bytes } => {
217                write!(f, "result text allocation of {bytes} bytes failed")
218            }
219            Self::ArithmeticOverflow => f.write_str("result materialization counter overflow"),
220            Self::Query(error) => error.fmt(f),
221            Self::Storage(error) => error.fmt(f),
222        }
223    }
224}
225impl std::error::Error for MaterializationError {
226    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
227        match self {
228            Self::Storage(error) => Some(error),
229            Self::Query(error) => Some(error),
230            _ => None,
231        }
232    }
233}
234
235/// Rank-addressed text access available only inside a Store query callback.
236pub struct QueryMaterializer<'a> {
237    #[cfg(any(test, feature = "test-seams"))]
238    pub(super) store: &'a Store,
239    pub(super) snapshot: &'a PublishedSnapshot,
240    pub(super) active: &'a ActiveSegment,
241    pub(super) cancellation: &'a QueryCancellation<'a>,
242    pub(super) address: &'a dyn Fn(usize) -> Result<ResultAddress, MaterializationError>,
243    pub(super) counts: Cell<crate::diag::MaterializationCounters>,
244}
245
246#[derive(Clone, Copy, Eq, PartialEq)]
247pub(super) struct ResultAddress {
248    pub(super) row_id: GlobalRowId,
249    pub(super) document: Option<DocumentVersion>,
250}
251
252pub(super) type HybridAddresses = super::stats::Accounted<Vec<(DocId, ResultAddress)>>;
253
254impl From<&SearchCandidate> for ResultAddress {
255    fn from(candidate: &SearchCandidate) -> Self {
256        Self {
257            row_id: candidate.row_id(),
258            document: candidate.document(),
259        }
260    }
261}
262
263impl QueryMaterializer<'_> {
264    /// Copies the exact stored text and version associated with one result rank.
265    pub fn text(&self, rank: usize) -> Result<MaterializedRow, MaterializationError> {
266        self.check()?;
267        let address = (self.address)(rank)?;
268        self.materialize_address(address)
269    }
270
271    /// Plants a wrong or missing ranked identity at the actual physical-row validation seam.
272    #[cfg(any(test, feature = "test-seams"))]
273    pub fn text_with_test_document(
274        &self,
275        rank: usize,
276        document: Option<DocumentVersion>,
277    ) -> Result<MaterializedRow, MaterializationError> {
278        self.check()?;
279        let mut address = (self.address)(rank)?;
280        address.document = document;
281        self.materialize_address(address)
282    }
283
284    fn materialize_address(
285        &self,
286        address: ResultAddress,
287    ) -> Result<MaterializedRow, MaterializationError> {
288        let row_id = address.row_id;
289        let document = address
290            .document
291            .ok_or(MaterializationError::MissingIdentity { row_id })?;
292        #[cfg(any(test, feature = "test-seams"))]
293        self.store
294            .text_materialization_work
295            .direct_row_lookups
296            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
297        let mut counts = self.counts.get();
298        counts.row_lookups = counts
299            .row_lookups
300            .checked_add(1)
301            .ok_or(MaterializationError::ArithmeticOverflow)?;
302        self.counts.set(counts);
303        let row = row_id.local_row() as usize;
304        let (actual, source) = match row_id.source() {
305            RowSource::Active => (
306                self.active.document(row),
307                self.active
308                    .text(row)
309                    .map_err(MaterializationError::Storage)?,
310            ),
311            RowSource::Sealed(id) => {
312                let segment = self
313                    .snapshot
314                    .segments()
315                    .iter()
316                    .find(|segment| segment.meta().id == id)
317                    .ok_or(MaterializationError::MissingSource { row_id })?;
318                let actual = segment
319                    .document_version(row)
320                    .map_err(StoreError::Segment)
321                    .map_err(MaterializationError::Storage)?;
322                let source = segment
323                    .query_stored_text()
324                    .map_err(StoreError::Segment)
325                    .map_err(MaterializationError::Storage)?
326                    .and_then(|text| text.row(row).flatten());
327                (actual, source)
328            }
329        };
330        require_document_version(document, actual).map_err(|mismatch| {
331            MaterializationError::IdentityMismatch {
332                row_id,
333                expected: mismatch.expected,
334                actual: mismatch.actual,
335            }
336        })?;
337        let source = source.ok_or(MaterializationError::MissingText { row_id, document })?;
338        let mut text = String::new();
339        text.try_reserve_exact(source.len()).map_err(|_| {
340            MaterializationError::AllocationFailed {
341                bytes: source.len(),
342            }
343        })?;
344        text.push_str(source);
345        let mut counts = self.counts.get();
346        counts.text_copies = counts
347            .text_copies
348            .checked_add(1)
349            .ok_or(MaterializationError::ArithmeticOverflow)?;
350        counts.text_bytes = counts
351            .text_bytes
352            .checked_add(source.len() as u64)
353            .ok_or(MaterializationError::ArithmeticOverflow)?;
354        self.counts.set(counts);
355        self.check()?;
356        Ok(MaterializedRow {
357            document,
358            row_id,
359            text,
360        })
361    }
362
363    fn check(&self) -> Result<(), MaterializationError> {
364        self.cancellation
365            .check_graph()
366            .map_err(super::map_scan_error)
367            .map_err(MaterializationError::Query)
368    }
369}
370
371impl Store {
372    /// Starts lexical retrieval before obtaining a caller-owned query vector.
373    /// Both legs and returned text use one admission pinned before the producer
374    /// runs. All submitted lexical work is joined on producer failure or panic.
375    ///
376    /// The returned vector borrow must remain valid for the complete query.
377    /// Producer errors retain their original type; core failures are separate.
378    ///
379    /// # Errors
380    /// Returns the producer's error or a typed admission/search failure.
381    pub fn search_hybrid_with_text_deferred<'vector, E, R>(
382        &self,
383        prepare_vector: impl FnOnce() -> Result<crate::ingest::SearchRequest<'vector>, E>,
384        lexical: &crate::fts::search::TermQuery,
385        query: &crate::fusion::HybridQuery,
386        options: impl Into<super::SearchOptions>,
387        control: super::QueryControl,
388        finish: impl FnOnce(&crate::ingest::StoreHybridSearchOutcome, &QueryMaterializer<'_>) -> R,
389    ) -> Result<(crate::ingest::StoreHybridSearchOutcome, R), HybridPreparationError<E>> {
390        self.search_hybrid_pinned_with_text(
391            prepare_vector,
392            None,
393            None,
394            super::PinnedLexicalQuery::Term(lexical),
395            query,
396            options.into(),
397            control,
398            finish,
399        )
400    }
401
402    #[allow(clippy::too_many_arguments)] // Pinned query inputs plus its scoped continuation.
403    fn search_hybrid_pinned_with_text<'vector, E, R>(
404        &self,
405        prepare_vector: impl FnOnce() -> Result<crate::ingest::SearchRequest<'vector>, E>,
406        filter: Option<&super::QueryFilter>,
407        eligible: Option<&[crate::ingest::DocId]>,
408        lexical: super::PinnedLexicalQuery<'_>,
409        query: &crate::fusion::HybridQuery,
410        options: super::SearchOptions,
411        control: super::QueryControl,
412        finish: impl FnOnce(&crate::ingest::StoreHybridSearchOutcome, &QueryMaterializer<'_>) -> R,
413    ) -> Result<(crate::ingest::StoreHybridSearchOutcome, R), HybridPreparationError<E>> {
414        self.search_hybrid_prepared_then(
415            prepare_vector,
416            filter,
417            eligible,
418            lexical,
419            query,
420            options,
421            control,
422            true,
423            None,
424            |mut outcome, snapshot, active, cancellation, addresses| {
425                cancellation.check_graph().map_err(super::map_scan_error)?;
426                let address = |rank| {
427                    let hit =
428                        outcome
429                            .hits
430                            .get(rank)
431                            .ok_or(MaterializationError::RankOutOfRange {
432                                rank,
433                                returned: outcome.hits.len(),
434                            })?;
435                    let addresses = addresses
436                        .ok_or(MaterializationError::MissingFusedIdentity { document: hit.key })?;
437                    let index = addresses
438                        .binary_search_by_key(&hit.key, |(key, _)| *key)
439                        .map_err(|_| MaterializationError::MissingFusedIdentity {
440                            document: hit.key,
441                        })?;
442                    addresses
443                        .get(index)
444                        .map(|(_, address)| *address)
445                        .ok_or(MaterializationError::MissingFusedIdentity { document: hit.key })
446                };
447                let materializer = QueryMaterializer {
448                    #[cfg(any(test, feature = "test-seams"))]
449                    store: self,
450                    snapshot,
451                    active,
452                    cancellation,
453                    address: &address,
454                    counts: Cell::new(Default::default()),
455                };
456                let value = finish(&outcome, &materializer);
457                let counts = materializer.counts.get();
458                outcome.diagnostics.materialization = Some(counts);
459                cancellation.check_graph().map_err(super::map_scan_error)?;
460                Ok((outcome, value))
461            },
462        )
463    }
464
465    /// Reads literal counters without resetting concurrent observations.
466    #[cfg(any(test, feature = "test-seams"))]
467    #[must_use]
468    pub fn query_materialization_test_counters(&self) -> MaterializationTestCounters {
469        use std::sync::atomic::Ordering::Relaxed;
470        MaterializationTestCounters {
471            admissions: self.text_materialization_work.admissions.load(Relaxed),
472            reverse_version_lookups: self
473                .text_materialization_work
474                .reverse_version_lookups
475                .load(Relaxed),
476            direct_row_lookups: self
477                .text_materialization_work
478                .direct_row_lookups
479                .load(Relaxed),
480            #[cfg(feature = "query-timing")]
481            lexical_admission_locks: self
482                .text_materialization_work
483                .lexical_admission_locks
484                .load(Relaxed),
485            #[cfg(feature = "query-timing")]
486            lexical_admission_lock_nanos: self
487                .text_materialization_work
488                .lexical_admission_lock_nanos
489                .load(Relaxed),
490        }
491    }
492    /// Runs hybrid fusion and constructs caller-owned results before releasing its snapshot.
493    pub fn search_hybrid_with_text<R>(
494        &self,
495        vector: crate::ingest::SearchRequest<'_>,
496        lexical: &crate::fts::search::TermQuery,
497        query: &crate::fusion::HybridQuery,
498        options: impl Into<super::SearchOptions>,
499        control: super::QueryControl,
500        finish: impl FnOnce(&crate::ingest::StoreHybridSearchOutcome, &QueryMaterializer<'_>) -> R,
501    ) -> Result<(crate::ingest::StoreHybridSearchOutcome, R), crate::fusion::FusionError> {
502        self.search_hybrid_pinned_with_text(
503            || Ok::<_, std::convert::Infallible>(vector),
504            vector.filter,
505            vector.eligible,
506            super::PinnedLexicalQuery::Term(lexical),
507            query,
508            options.into(),
509            control,
510            finish,
511        )
512        .map_err(|error| match error {
513            HybridPreparationError::Preparation(never) => match never {},
514            HybridPreparationError::Search(error) => error,
515        })
516    }
517
518    /// Runs a flat-term lexical query and returns one snippet per candidate,
519    /// in rank order, cut from the text pinned for the query's generation.
520    ///
521    /// Highlights are the tokens the store's own analyzer produces from that
522    /// text whose analyzed term is one of the query's terms, which is exactly
523    /// what the lexical leg scored.
524    ///
525    /// # Errors
526    /// Returns the search failure, or [`crate::ingest::StoreLexicalError::MissingSnippetMatch`]
527    /// when a ranked row's stored text contains none of the query's terms.
528    pub fn search_lexical_with_snippets(
529        &self,
530        query: &crate::fts::search::TermQuery,
531        k: usize,
532        window: NonZeroUsize,
533        control: super::QueryControl,
534    ) -> Result<
535        (
536            crate::ingest::StoreLexicalSearchOutcome,
537            Vec<crate::fts::query::OwnedLexicalSnippet>,
538        ),
539        crate::ingest::StoreLexicalError,
540    > {
541        self.search_lexical_filtered_with_snippets(query, k, window, control, None)
542    }
543
544    /// Searches eligible lexical rows and materializes snippets at the same generation.
545    pub fn search_lexical_filtered_with_snippets(
546        &self,
547        query: &crate::fts::search::TermQuery,
548        k: usize,
549        window: NonZeroUsize,
550        control: super::QueryControl,
551        filter: Option<&super::QueryFilter>,
552    ) -> Result<
553        (
554            crate::ingest::StoreLexicalSearchOutcome,
555            Vec<crate::fts::query::OwnedLexicalSnippet>,
556        ),
557        crate::ingest::StoreLexicalError,
558    > {
559        self.search_lexical_then(
560            query,
561            k,
562            control,
563            filter,
564            |outcome, snapshot, active, assembly, hits, cancellation| {
565                let mut snippets = Vec::with_capacity(hits.len());
566                for hit in hits {
567                    cancellation.check_graph().map_err(super::map_scan_error)?;
568                    let (text, _) = super::structured_lexical_row(
569                        snapshot,
570                        active,
571                        &assembly.sources,
572                        hit.doc,
573                    )?;
574                    snippets.push(
575                        crate::fts::snippet::owned_window(
576                            &self.tokenizer,
577                            text,
578                            &query.terms,
579                            window,
580                        )
581                        .ok_or(
582                            crate::ingest::StoreLexicalError::MissingSnippetMatch {
583                                segment: hit.doc.segment,
584                                row: hit.doc.row,
585                            },
586                        )?,
587                    );
588                }
589                Ok((outcome, snippets))
590            },
591        )
592    }
593
594    /// Runs hybrid fusion over a flat-term lexical leg and returns one
595    /// optional snippet per fused hit, in rank order, cut from the text pinned
596    /// for the query's generation. A hit gets no snippet when its row has no
597    /// stored text or the text contains none of the query's terms.
598    ///
599    /// # Errors
600    /// Returns the search failure; the inner result carries a failure to read
601    /// a ranked row's text.
602    pub fn search_hybrid_with_snippets(
603        &self,
604        vector: crate::ingest::SearchRequest<'_>,
605        lexical: &crate::fts::search::TermQuery,
606        query: &crate::fusion::HybridQuery,
607        options: impl Into<super::SearchOptions>,
608        control: super::QueryControl,
609        window: NonZeroUsize,
610    ) -> Result<(crate::ingest::StoreHybridSearchOutcome, HybridSnippets), crate::fusion::FusionError>
611    {
612        self.hybrid_snippets(
613            vector,
614            super::PinnedLexicalQuery::Term(lexical),
615            query,
616            options.into(),
617            control,
618            window,
619        )
620    }
621
622    /// [`Self::search_hybrid_with_snippets`] for a structured lexical leg.
623    /// Highlights mark the terms of the leg's reported expansions, so a prefix
624    /// marks every indexed term it expanded to.
625    ///
626    /// # Errors
627    /// As [`Self::search_hybrid_with_snippets`].
628    pub fn search_hybrid_structured_with_snippets(
629        &self,
630        vector: crate::ingest::SearchRequest<'_>,
631        lexical: &crate::fts::query::LexicalQuery,
632        query: &crate::fusion::HybridQuery,
633        options: impl Into<super::SearchOptions>,
634        control: super::QueryControl,
635        window: NonZeroUsize,
636    ) -> Result<(crate::ingest::StoreHybridSearchOutcome, HybridSnippets), crate::fusion::FusionError>
637    {
638        self.hybrid_snippets(
639            vector,
640            super::PinnedLexicalQuery::Structured(lexical),
641            query,
642            options.into(),
643            control,
644            window,
645        )
646    }
647
648    fn hybrid_snippets(
649        &self,
650        vector: crate::ingest::SearchRequest<'_>,
651        lexical: super::PinnedLexicalQuery<'_>,
652        query: &crate::fusion::HybridQuery,
653        options: super::SearchOptions,
654        control: super::QueryControl,
655        window: NonZeroUsize,
656    ) -> Result<(crate::ingest::StoreHybridSearchOutcome, HybridSnippets), crate::fusion::FusionError>
657    {
658        let flat_terms = match lexical {
659            super::PinnedLexicalQuery::Term(query) => Some(&query.terms),
660            super::PinnedLexicalQuery::Structured(_) => None,
661        };
662        self.search_hybrid_pinned_with_text(
663            || Ok::<_, std::convert::Infallible>(vector),
664            vector.filter,
665            vector.eligible,
666            lexical,
667            query,
668            options,
669            control,
670            |outcome, materializer| {
671                let expanded;
672                let terms: &[Vec<u8>] = match flat_terms {
673                    Some(terms) => terms,
674                    None => {
675                        expanded = outcome
676                            .lexical_expansions
677                            .iter()
678                            .map(|expansion| expansion.term.clone())
679                            .collect::<Vec<_>>();
680                        &expanded
681                    }
682                };
683                let mut snippets = Vec::with_capacity(outcome.hits.len());
684                for rank in 0..outcome.hits.len() {
685                    let row = match materializer.text(rank) {
686                        Ok(row) => row,
687                        Err(MaterializationError::MissingText { .. }) => {
688                            snippets.push(None);
689                            continue;
690                        }
691                        Err(error) => return Err(error),
692                    };
693                    snippets.push(crate::fts::snippet::owned_window(
694                        &self.tokenizer,
695                        &row.text,
696                        terms,
697                        window,
698                    ));
699                }
700                Ok(snippets)
701            },
702        )
703        .map_err(|error| match error {
704            HybridPreparationError::Preparation(never) => match never {},
705            HybridPreparationError::Search(error) => error,
706        })
707    }
708
709    /// Runs a lexical query and constructs caller-owned results before releasing its snapshot.
710    pub fn search_lexical_with_text<R>(
711        &self,
712        query: &crate::fts::search::TermQuery,
713        k: usize,
714        control: super::QueryControl,
715        finish: impl FnOnce(&crate::ingest::StoreLexicalSearchOutcome, &QueryMaterializer<'_>) -> R,
716    ) -> Result<(crate::ingest::StoreLexicalSearchOutcome, R), crate::ingest::StoreLexicalError>
717    {
718        self.search_lexical_then(
719            query,
720            k,
721            control,
722            None,
723            |mut outcome, snapshot, active, assembly, hits, cancellation| {
724                cancellation.check_graph().map_err(super::map_scan_error)?;
725                let address = |rank| {
726                    let candidate = outcome.candidates.get(rank).ok_or(
727                        MaterializationError::RankOutOfRange {
728                            rank,
729                            returned: outcome.candidates.len(),
730                        },
731                    )?;
732                    let hit = hits.get(rank).ok_or(MaterializationError::RankOutOfRange {
733                        rank,
734                        returned: hits.len(),
735                    })?;
736                    Ok(ResultAddress {
737                        row_id: lexical_row_id(snapshot, &assembly.sources, hit.doc)?,
738                        document: Some(candidate.document),
739                    })
740                };
741                let materializer = QueryMaterializer {
742                    #[cfg(any(test, feature = "test-seams"))]
743                    store: self,
744                    snapshot,
745                    active,
746                    cancellation,
747                    address: &address,
748                    counts: Cell::new(Default::default()),
749                };
750                let value = finish(&outcome, &materializer);
751                let counts = materializer.counts.get();
752                outcome.diagnostics.materialization = Some(counts);
753                cancellation.check_graph().map_err(super::map_scan_error)?;
754                Ok((outcome, value))
755            },
756        )
757    }
758
759    /// Runs a dense query and constructs owned caller results through rank-addressed text access.
760    pub fn search_with_text<R>(
761        &self,
762        request: crate::ingest::SearchRequest<'_>,
763        k: usize,
764        options: impl Into<super::SearchOptions>,
765        control: super::QueryControl,
766        finish: impl FnOnce(&crate::ingest::SearchOutcome, &QueryMaterializer<'_>) -> R,
767    ) -> Result<(crate::ingest::SearchOutcome, R), super::QueryError> {
768        self.search_with_graph_bound_mode_then(
769            request,
770            k,
771            options.into(),
772            control,
773            super::GraphBoundMode::Shared,
774            |mut outcome, admitted, control| {
775                let lease =
776                    SnapshotLease::new_at(Arc::clone(&admitted.snapshot), admitted.generation);
777                let cancellation = QueryCancellation::new(control, &lease);
778                cancellation.check_graph().map_err(super::map_scan_error)?;
779                let address = |rank| {
780                    outcome.candidates.get(rank).map(ResultAddress::from).ok_or(
781                        MaterializationError::RankOutOfRange {
782                            rank,
783                            returned: outcome.candidates.len(),
784                        },
785                    )
786                };
787                let materializer = QueryMaterializer {
788                    #[cfg(any(test, feature = "test-seams"))]
789                    store: self,
790                    counts: Cell::new(Default::default()),
791                    snapshot: &admitted.snapshot,
792                    active: &admitted.active_segment,
793                    cancellation: &cancellation,
794                    address: &address,
795                };
796                let value = finish(&outcome, &materializer);
797                let counts = materializer.counts.get();
798                outcome.diagnostics.materialization = Some(counts);
799                cancellation.check_graph().map_err(super::map_scan_error)?;
800                Ok((outcome, value))
801            },
802        )
803    }
804}
805
806fn lexical_row_id(
807    snapshot: &PublishedSnapshot,
808    sources: &[super::StructuredLexicalSource],
809    doc: crate::fts::search::GlobalDocId,
810) -> Result<GlobalRowId, MaterializationError> {
811    let source =
812        sources
813            .get(doc.segment as usize)
814            .ok_or(MaterializationError::InvalidLexicalSource {
815                source: doc.segment,
816            })?;
817    let source = match source {
818        super::StructuredLexicalSource::Active => RowSource::Active,
819        super::StructuredLexicalSource::Sealed(ordinal) => RowSource::Sealed(
820            snapshot
821                .segments()
822                .get(*ordinal)
823                .ok_or(MaterializationError::InvalidLexicalSource {
824                    source: doc.segment,
825                })?
826                .meta()
827                .id,
828        ),
829    };
830    Ok(GlobalRowId::new(source, doc.row))
831}
zeppelin_embed/lifecycle/materialize.rs

zeppelin_embed/lifecycle/
materialize.rs
