kevy_embedded/
ops_index_cold.rs

1//! The cold-aware halves of the scalar query surface: the windowed
2//! per-shard iterator and the range/count entry points that merge a
3//! windowed index's cold segments into the hot tree's answer. Split
4//! from `ops_index.rs` for the 500-LOC house rule.
5
6use kevy_index::{Cursor, IndexSpec, IndexValue, Segment};
7
8use crate::ops_index::{IndexPage, WinRef, merge_page, sync_segs};
9use crate::store::{Store, lock_write};
10use crate::{KevyError, KevyResult};
11
12impl Store {
13    /// Range / EQ query with cursor pagination: merged across shards
14    /// in `(value, key)` order — including any cold segments a
15    /// windowed index has slid out. `cursor = None` starts; the
16    /// returned cursor resumes exclusively.
17    pub fn idx_query(
18        &self,
19        name: &[u8],
20        min: &IndexValue,
21        max: &IndexValue,
22        cursor: Option<&Cursor>,
23        limit: usize,
24    ) -> KevyResult<IndexPage> {
25        let r = self.idx_query_gather(name, min, max, cursor, limit);
26        self.observe_noindex(name, kevy_index::AdviseShape::Range, &r);
27        if r.is_ok() {
28            self.observe_hit(name);
29        }
30        r
31    }
32
33    fn idx_query_gather(
34        &self,
35        name: &[u8],
36        min: &IndexValue,
37        max: &IndexValue,
38        cursor: Option<&Cursor>,
39        limit: usize,
40    ) -> KevyResult<IndexPage> {
41        let limit = limit.clamp(1, 100_000);
42        let mut all: Vec<(IndexValue, Vec<u8>)> = Vec::new();
43        #[cfg(not(target_arch = "wasm32"))]
44        let probe = self.usage_cell(name);
45        self.for_each_segment_windowed(name, |spec, seg, win| {
46            let (hits, _) = seg.range(min, max, cursor, limit);
47            all.extend(hits.into_iter().map(|(k, v)| (v, k)));
48            #[cfg(not(target_arch = "wasm32"))]
49            if let Some(w) = win {
50                crate::ops_index::advise::probe_window(&probe, w, min);
51            }
52            #[cfg(not(target_arch = "wasm32"))]
53            if let Some(w) = win.filter(|w| w.has_cold()) {
54                // The cursor goes INTO the cold walk so the limit
55                // counts post-cursor entries — filtering afterwards
56                // starves the cold side on any page after the first
57                // (its limit fills with pre-cursor entries that all
58                // drop).
59                let cold = w
60                    .cold_hits(spec.ty, min, max, cursor, limit)
61                    .map_err(|e| KevyError::Io(std::io::Error::other(e)))?;
62                all.extend(cold.into_iter().map(|(k, v)| (v, k)));
63            }
64            #[cfg(target_arch = "wasm32")]
65            let _ = (spec, win);
66            Ok(())
67        })?;
68        Ok(merge_page(all, limit))
69    }
70
71    /// Count without materializing keys — hot tree plus cold segments.
72    pub fn idx_count(&self, name: &[u8], min: &IndexValue, max: &IndexValue) -> KevyResult<u64> {
73        let r = self.idx_count_gather(name, min, max);
74        self.observe_noindex(name, kevy_index::AdviseShape::Range, &r);
75        if r.is_ok() {
76            self.observe_hit(name);
77        }
78        r
79    }
80
81    fn idx_count_gather(&self, name: &[u8], min: &IndexValue, max: &IndexValue) -> KevyResult<u64> {
82        let mut total = 0u64;
83        #[cfg(not(target_arch = "wasm32"))]
84        let probe = self.usage_cell(name);
85        self.for_each_segment_windowed(name, |spec, seg, win| {
86            total += seg.count(min, max);
87            #[cfg(not(target_arch = "wasm32"))]
88            if let Some(w) = win {
89                crate::ops_index::advise::probe_window(&probe, w, min);
90            }
91            #[cfg(not(target_arch = "wasm32"))]
92            if let Some(w) = win.filter(|w| w.has_cold()) {
93                total += w
94                    .cold_count(spec.ty, min, max)
95                    .map_err(|e| KevyError::Io(std::io::Error::other(e)))?;
96            }
97            #[cfg(target_arch = "wasm32")]
98            let _ = (spec, win);
99            Ok(())
100        })?;
101        Ok(total)
102    }
103
104
105    /// [`Self::for_each_segment`], with each shard's window runtime
106    /// (if any) beside the segment, and a fallible visitor — the cold
107    /// half does I/O, and a corrupt cold segment must become the
108    /// query's error, never a partial answer.
109    pub(crate) fn for_each_segment_windowed(
110        &self,
111        name: &[u8],
112        mut f: impl FnMut(&IndexSpec, &Segment, WinRef<'_>) -> KevyResult<()>,
113    ) -> KevyResult<()> {
114        let mut found = false;
115        for shard in self.shards.iter() {
116            let mut g = lock_write(shard);
117            let inner = &mut *g;
118            sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
119            let segs = &inner.idx_segs;
120            if let Some((spec, seg)) = segs.segs.iter().find(|(s, _)| s.name == name) {
121                found = true;
122                #[cfg(not(target_arch = "wasm32"))]
123                let win = segs.window_of(name);
124                #[cfg(target_arch = "wasm32")]
125                let win = None;
126                f(spec, seg, win)?;
127            }
128        }
129        if found {
130            Ok(())
131        } else {
132            Err(KevyError::NotFound("no such index".into()))
133        }
134    }
135
136}
kevy_embedded/ops_index_cold.rs

kevy_embedded/
ops_index_cold.rs
