rac_engine/
read_model.rs

1//! Store-served search (ADR-104) — `decided find` answered from the memory-
2//! mapped base, byte-identical to the fresh walk (INDEX-PLAN B3).
3//!
4//! The candidate set comes from the term-major postings and only candidates
5//! are reconstructed; the corpus-global statistics the scorer needs — n and
6//! the per-field Σ from the header, document frequency from the prefix
7//! ranges — carry the non-matching corpus's contribution without touching
8//! its rows. Matching, snippets, and ranking run through the one shared
9//! `rank_and_build` tail, so warm bytes equal cold bytes by construction.
10//!
11//! Deliberate divergence from the oracle's warm path (PORT-CONTRACT.d/10
12//! §0a): a query term listed twice contributes its document frequency once
13//! per OCCURRENCE here, exactly as the fresh walk counts it. The oracle's
14//! store path dedups (an ADR-112 violation recorded as an oracle defect);
15//! the native engine keeps warm == cold instead.
16
17use crate::index_store::MmapIndexReader;
18use crate::resolve::{
19    entry_has_tags, entry_is_retired, match_entry_with_fields, rank_and_build, tokenize,
20    CorpusStats, SearchResult,
21};
22
23/// `decided find` served from the store — reproduces `search_index_filtered`.
24pub fn store_search(
25    reader: &MmapIndexReader,
26    query: &str,
27    artifact_type: Option<&str>,
28    tags: &[String],
29    live_only: bool,
30) -> SearchResult {
31    let timing = crate::timing::enabled();
32    let tokenize_started = timing.then(std::time::Instant::now);
33    let terms = tokenize(query);
34    if let Some(started) = tokenize_started {
35        crate::timing::emit(
36            "search.query_tokenize",
37            started.elapsed(),
38            &[("terms", terms.len() as u64)],
39        );
40    }
41    let empty = || SearchResult {
42        query: query.to_string(),
43        artifact_type: artifact_type.map(str::to_string),
44        matches: Vec::new(),
45    };
46    if terms.is_empty() {
47        return empty();
48    }
49    let tag_filter: Vec<String> = tags.iter().map(|t| crate::pycompat::py_casefold(t)).collect();
50
51    // AND matching requires every distinct term somewhere in the document,
52    // so only the intersection of their cross-field postings can match. Keep
53    // the set ascending so matched rows retain walk (docid) order.
54    let mut postings = Vec::new();
55    let mut distinct = std::collections::HashSet::new();
56    let mut postings_duration = std::time::Duration::ZERO;
57    let mut merge_duration = std::time::Duration::ZERO;
58    for term in &terms {
59        if !distinct.insert(term.as_str()) {
60            continue;
61        }
62        let started = timing.then(std::time::Instant::now);
63        let decoded = reader.prefix_docids(term);
64        if let Some(started) = started {
65            postings_duration += started.elapsed();
66        }
67        match decoded {
68            Ok(docids) => postings.push(docids),
69            Err(_) => return empty(), // corrupt row mid-read: valid empty, never a crash
70        }
71    }
72    postings.sort_by_key(std::collections::BTreeSet::len);
73    let started = timing.then(std::time::Instant::now);
74    let mut postings = postings.into_iter();
75    let mut candidates = postings.next().unwrap_or_default();
76    for docids in postings {
77        candidates.retain(|docid| docids.contains(docid));
78        if candidates.is_empty() {
79            break;
80        }
81    }
82    if let Some(started) = started {
83        merge_duration += started.elapsed();
84    }
85    crate::timing::emit(
86        "search.postings_decode",
87        postings_duration,
88        &[("terms", terms.len() as u64)],
89    );
90    crate::timing::emit(
91        "search.candidate_merge",
92        merge_duration,
93        &[("candidates", candidates.len() as u64)],
94    );
95
96    let mut matched = Vec::new();
97    let candidate_count = candidates.len() as u64;
98    let mut row_decode_duration = std::time::Duration::ZERO;
99    let mut row_tokenize_duration = std::time::Duration::ZERO;
100    let mut matching_duration = std::time::Duration::ZERO;
101    for docid in candidates {
102        let started = timing.then(std::time::Instant::now);
103        let decoded = reader.full_entry(docid);
104        if let Some(started) = started {
105            row_decode_duration += started.elapsed();
106        }
107        let Ok(entry) = decoded else {
108            continue;
109        };
110        if let Some(t) = artifact_type {
111            if entry.artifact_type != t {
112                continue;
113            }
114        }
115        if !tag_filter.is_empty() && !entry_has_tags(&entry, &tag_filter) {
116            continue;
117        }
118        let started = timing.then(std::time::Instant::now);
119        let decoded_fields = reader.field_tokens(docid);
120        if let Some(started) = started {
121            row_tokenize_duration += started.elapsed();
122        }
123        let Ok(fields) = decoded_fields else {
124            continue;
125        };
126        let started = timing.then(std::time::Instant::now);
127        let matched_entry = match_entry_with_fields(&entry, &fields, &terms);
128        if let Some(started) = started {
129            matching_duration += started.elapsed();
130        }
131        if let Some(m) = matched_entry {
132            matched.push((entry, fields, m));
133        }
134    }
135    crate::timing::emit(
136        "search.row_decode",
137        row_decode_duration,
138        &[("candidates", candidate_count)],
139    );
140    crate::timing::emit(
141        "search.row_tokenize",
142        row_tokenize_duration,
143        &[("candidates", candidate_count)],
144    );
145    crate::timing::emit(
146        "search.matching",
147        matching_duration,
148        &[("matched", matched.len() as u64)],
149    );
150    if live_only && !matched.is_empty() {
151        matched.retain(|(entry, _, _)| !entry_is_retired(entry));
152    }
153    if matched.is_empty() {
154        return empty();
155    }
156
157    // Corpus-global statistics from the store's integer accumulators: the
158    // same values the walk derives, without materialising unmatched rows.
159    let n = i64::from(reader.doc_count);
160    let mut avglen = [0.0f64; 6];
161    for (i, sum) in reader.field_length_sums.iter().enumerate() {
162        avglen[i] = if n != 0 { *sum as f64 / n as f64 } else { 0.0 };
163    }
164    let mut df: std::collections::HashMap<String, i64> = std::collections::HashMap::new();
165    let mut occurrences: std::collections::HashMap<&str, i64> = std::collections::HashMap::new();
166    for term in &terms {
167        *occurrences.entry(term.as_str()).or_insert(0) += 1;
168    }
169    for (term, occ) in occurrences {
170        let doc_count = reader
171            .prefix_docids(term)
172            .map(|d| d.len() as i64)
173            .unwrap_or(0);
174        // Per-OCCURRENCE df, the fresh walk's accounting (see module doc).
175        df.insert(term.to_string(), occ * doc_count);
176    }
177    let stats = CorpusStats { n, df, avglen };
178
179    let scored: Vec<_> = matched
180        .iter()
181        .map(|(entry, fields, m)| (entry, fields, m.clone()))
182        .collect();
183    rank_and_build(query, artifact_type, scored, &terms, &stats)
184}
185
186/// Live-decision topic search served from the store (ADR-067): the
187/// decision-typed search, then the liveness filter over the precomputed
188/// live-decision paths — `ReadModelView.find_decisions`.
189pub fn store_find_decisions(reader: &MmapIndexReader, topic: &str) -> SearchResult {
190    let mut result = store_search(reader, topic, Some("decision"), &[], false);
191    let live: std::collections::HashSet<String> =
192        reader.live_decision_paths().unwrap_or_default().into_iter().collect();
193    result.matches.retain(|m| live.contains(&m.path));
194    result
195}
196
197/// Point resolution over the persisted alias map — `Fold.resolve`,
198/// byte-identical to `resolve_in_index` over a walk of the same corpus.
199pub fn store_resolve(
200    reader: &MmapIndexReader,
201    artifact_id: &str,
202) -> crate::resolve::ResolutionResult {
203    use crate::resolve::{OUTCOME_DUPLICATE, OUTCOME_NOT_FOUND, OUTCOME_RESOLVED};
204    let wanted =
205        crate::pycompat::py_casefold(crate::pycompat::py_strip(artifact_id));
206    let docids = reader.alias_docids(&wanted).unwrap_or_default();
207    if docids.is_empty() {
208        return crate::resolve::ResolutionResult {
209            artifact_id: artifact_id.to_string(),
210            outcome: OUTCOME_NOT_FOUND,
211            artifact: None,
212            duplicate_paths: Vec::new(),
213        };
214    }
215    if docids.len() > 1 {
216        let mut paths: Vec<String> = docids
217            .iter()
218            .filter_map(|&docid| reader.entry_path(docid).ok())
219            .collect();
220        paths.sort();
221        return crate::resolve::ResolutionResult {
222            artifact_id: artifact_id.to_string(),
223            outcome: OUTCOME_DUPLICATE,
224            artifact: None,
225            duplicate_paths: paths,
226        };
227    }
228    match reader.identity_entry(docids[0]) {
229        // `from_entry` copies whatever tags the resolved projection carries:
230        // the store's identity rows DO persist tags (ADR-109), so the mapped
231        // base resolves WITH them — exactly as the oracle's Fold does. (The
232        // delta snapshot resolves over the tag-free identity projection; that
233        // asymmetry is the oracle's, mirrored at the caller.)
234        Ok(entry) => crate::resolve::ResolutionResult {
235            artifact_id: artifact_id.to_string(),
236            outcome: OUTCOME_RESOLVED,
237            artifact: Some(crate::resolve::resolved_from_entry(&entry)),
238            duplicate_paths: Vec::new(),
239        },
240        Err(_) => crate::resolve::ResolutionResult {
241            artifact_id: artifact_id.to_string(),
242            outcome: OUTCOME_NOT_FOUND,
243            artifact: None,
244            duplicate_paths: Vec::new(),
245        },
246    }
247}
248
249/// Every identity row of the mapped base, in docid (walk) order — the
250/// materialised projection `get_related`'s graph helpers read.
251pub fn store_identity_entries(
252    reader: &MmapIndexReader,
253) -> Vec<crate::resolve::IndexEntry> {
254    (0..reader.doc_count)
255        .filter_map(|docid| reader.identity_entry(docid).ok())
256        .collect()
257}
258
259/// `find_decisions_in` over already-derived structures — the fresh-build
260/// arm of the cache seam (`_find_from_store`'s `else` branch).
261pub fn find_decisions_in(
262    entries: &[crate::resolve::IndexEntry],
263    live_paths: &[String],
264    topic: &str,
265) -> SearchResult {
266    let mut result = crate::resolve::search_index(entries, topic, Some("decision"), &[]);
267    let live: std::collections::HashSet<&str> = live_paths.iter().map(String::as_str).collect();
268    result.matches.retain(|m| live.contains(m.path.as_str()));
269    result
270}
rac_engine/read_model.rs

rac_engine/
read_model.rs
