kevy_embedded/
ops_view.rs

1//! Embedded views (server parity minus VIA/FIELDS hydration:
2//! in-process callers dereference and read fields directly).
3//!
4//! Mirrors the embedded index architecture: per-shard view states in
5//! `Inner` (maintained inside `commit_write` right after index
6//! maintenance, under the same shard lock), a store-level registry,
7//! synchronous builds, typed API (`Tree` passed directly — no text
8//! grammar in-process).
9
10use crate::{KevyError, KevyResult};
11use std::io;
12use std::sync::RwLock;
13
14use kevy_index::{
15    IndexValue, MaterializedSet, Membership, SortOrder, Tree, ViewCatalog, ViewMode, ViewSpec,
16};
17
18use crate::ops_index::ShardSegs;
19use crate::store::{Store, lock_write};
20
21/// Store-level registry.
22#[derive(Debug, Default)]
23pub(crate) struct ViewReg {
24    pub(crate) catalog: RwLock<(u64, ViewCatalog)>,
25}
26
27/// One shard's view states (inside `Inner`, guarded by the shard lock).
28#[derive(Debug, Default)]
29pub(crate) struct ShardViews {
30    pub(crate) version: u64,
31    /// The (view catalog, index list) versions the key directories were
32    /// last set for.
33    pub(crate) dirs_at: (u64, u64),
34    pub(crate) views: Vec<ViewState>,
35    /// `reserved_bytes` generation cache — see
36    /// `ShardSegs::stats_dirty`; same contract, view half. Tier-only,
37    /// like its twin.
38    #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
39    pub(crate) stats_dirty: bool,
40    #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
41    pub(crate) reserved_cache: u64,
42}
43
44#[derive(Debug)]
45
46pub(crate) struct ViewState {
47    spec: ViewSpec,
48    mat: Option<MaterializedSet>,
49    needs_rebuild: bool,
50}
51
52impl ShardViews {
53    /// Invalidate the cache — no-op without the tier backend; see
54    /// `ShardSegs::mark_stats_dirty`.
55    #[inline]
56    pub(crate) fn mark_stats_dirty(&mut self) {
57        #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
58        {
59            self.stats_dirty = true;
60        }
61    }
62
63    /// Σ approximate heap bytes of the materialized view sets — the
64    /// view half of the tier's `reserved_bytes` feed.
65    /// Virtual views hold no set and contribute nothing.
66    #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
67    pub(crate) fn reserved_bytes(&mut self) -> u64 {
68        if !self.stats_dirty {
69            return self.reserved_cache;
70        }
71        let sum = self
72            .views
73            .iter()
74            .map(|v| v.mat.as_ref().map_or(0, MaterializedSet::approx_bytes))
75            .sum();
76        self.reserved_cache = sum;
77        self.stats_dirty = false;
78        sum
79    }
80}
81
82/// One page of view members plus the resume cursor.
83///
84/// ```
85/// use kevy_embedded::*;
86/// let s = Store::open(Config::default())?;
87/// s.idx_create(b"by_pri", b"t:", b"pri", IndexValType::I64, IndexKind::Range)?;
88/// for (k, pri) in [(&b"t:1"[..], &b"5"[..]), (b"t:2", b"9")] {
89///     s.hset(k, &[(b"pri", pri)])?;
90/// }
91/// let all = ViewTree::Leaf(ViewLeaf::new(b"by_pri".to_vec(), IndexValue::I64(0), IndexValue::I64(99)));
92/// s.view_create(b"urgent", all, b"by_pri", SortOrder::Desc, ViewMode::Virtual)?;
93/// let (members, after): ViewPage = s.view_query(b"urgent", None, 1)?;
94/// assert_eq!(members, [(b"t:2".to_vec(), IndexValue::I64(9))]);
95/// let (rest, _) = s.view_query(b"urgent", after.as_ref(), 10)?; // resume past it
96/// assert_eq!(rest[0].0, b"t:1");
97/// # Ok::<(), kevy_embedded::KevyError>(())
98/// ```
99pub type ViewPage = (Vec<(Vec<u8>, IndexValue)>, Option<(IndexValue, Vec<u8>)>);
100
101impl Store {
102    /// Declare a view (typed tree; `via` is not supported embedded —
103    /// read fields in-process). Builds synchronously.
104    pub fn view_create(
105        &self,
106        name: &[u8],
107        tree: Tree,
108        order_by: &[u8],
109        order: SortOrder,
110        mode: ViewMode,
111    ) -> KevyResult<()> {
112        self.catalog_change(|| self.create_view(name, tree, order_by, order, mode))
113    }
114
115    fn create_view(
116        &self,
117        name: &[u8],
118        tree: Tree,
119        order_by: &[u8],
120        order: SortOrder,
121        mode: ViewMode,
122    ) -> KevyResult<()> {
123        self.check_view_refs(&tree, order_by)?;
124        let spec = ViewSpec::new(name, tree, order_by).with_order(order).with_mode(mode);
125        {
126            let mut g =
127                self.views.catalog.write().unwrap_or_else(std::sync::PoisonError::into_inner);
128            let (ver, cat) = &mut *g;
129            cat.create(spec).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
130            *ver += 1;
131        }
132        for shard in self.shards.iter() {
133            let mut g = lock_write(shard);
134            let inner = &mut *g;
135            crate::ops_index::sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
136            sync_views(&self.views, &mut inner.view_segs, &mut inner.idx_segs);
137        }
138        Ok(())
139    }
140
141    /// Every index a view references (its leaves + ORDER BY) must
142    /// already be declared.
143    fn check_view_refs(&self, tree: &Tree, order_by: &[u8]) -> KevyResult<()> {
144        let mut names: Vec<Vec<u8>> = vec![order_by.to_vec()];
145        tree.each_leaf(&mut |l| names.push(l.index.clone()));
146        let g = self.indexes.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
147        for n in &names {
148            if g.1.get(n).is_none() {
149                return Err(KevyError::InvalidInput("view references unknown index".into()));
150            }
151        }
152        Ok(())
153    }
154
155    /// Drop a view; `false` if absent. Refused on a replica and after
156    /// [`Store::shutdown`], like every write.
157    pub fn view_drop(&self, name: &[u8]) -> KevyResult<bool> {
158        self.catalog_change(|| {
159            let mut g =
160                self.views.catalog.write().unwrap_or_else(std::sync::PoisonError::into_inner);
161            let (ver, cat) = &mut *g;
162            let hit = cat.drop_view(name);
163            if hit {
164                *ver += 1;
165            }
166            Ok(hit)
167        })
168    }
169
170    /// Ordered page across shards (`after` resumes exclusively; DESC
171    /// views page from the large end).
172    pub fn view_query(
173        &self,
174        name: &[u8],
175        after: Option<&(IndexValue, Vec<u8>)>,
176        limit: usize,
177    ) -> KevyResult<ViewPage> {
178        let limit = limit.clamp(1, 100_000);
179        let mut desc = false;
180        let mut all: Vec<(IndexValue, Vec<u8>)> = Vec::new();
181        let mut found = false;
182        for shard in self.shards.iter() {
183            let mut g = lock_write(shard);
184            let inner = &mut *g;
185            crate::ops_index::sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
186            sync_views(&self.views, &mut inner.view_segs, &mut inner.idx_segs);
187            let Some(i) = inner.view_segs.views.iter().position(|v| v.spec.name == name) else {
188                continue;
189            };
190            found = true;
191            if inner.view_segs.views[i].needs_rebuild {
192                inner.view_segs.mark_stats_dirty();
193                rebuild(&mut inner.view_segs.views[i], &inner.idx_segs);
194            }
195            let vs = &inner.view_segs.views[i];
196            desc = vs.spec.order == SortOrder::Desc;
197            match &vs.mat {
198                Some(m) => all.extend(m.page(after, limit)),
199                None => stream_virtual(&vs.spec, &inner.idx_segs, after, limit, &mut all),
200            }
201        }
202        if !found {
203            return Err(KevyError::NotFound("no such view".into()));
204        }
205        all.sort();
206        if desc {
207            all.reverse();
208        }
209        all.truncate(limit);
210        let next = if all.len() == limit { all.last().cloned() } else { None };
211        Ok((all.into_iter().map(|(v, k)| (k, v)).collect(), next))
212    }
213
214    /// Declared views (name, mode, leaves).
215    pub fn view_list(&self) -> Vec<(Vec<u8>, ViewMode, usize)> {
216        let g = self.views.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
217        g.1.iter().map(|s| (s.name.clone(), s.mode, s.tree.leaves())).collect()
218    }
219
220    /// The declaration of the view named `name`, as the catalog holds it.
221    pub fn view_spec(&self, name: &[u8]) -> Option<kevy_index::ViewSpec> {
222        let g = self.views.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
223        g.1.get(name).cloned()
224    }
225
226    /// Summed member count across shards.
227    pub fn view_count(&self, name: &[u8]) -> KevyResult<u64> {
228        Ok(self.view_query(name, None, 100_000)?.0.len() as u64)
229    }
230}
231
232fn resolver<'a>(segs: &'a ShardSegs) -> impl Fn(&[u8]) -> Option<&'a kevy_index::Segment> {
233    move |name: &[u8]| segs.segs.iter().find(|(s, _)| s.name() == name).map(|(_, seg)| seg)
234}
235
236fn eval_shard(spec: &ViewSpec, segs: &ShardSegs) -> Vec<(IndexValue, Vec<u8>)> {
237    let r = resolver(segs);
238    let members = spec.tree.eval(&&r);
239    members
240        .into_iter()
241        .filter_map(|k| r(&spec.order_by).and_then(|s| s.key_dir()?.get(&k)).map(|v| (v, k)))
242        .collect()
243}
244
245/// Virtual-mode page: order-driven streaming over the ORDER BY index,
246/// probing the tree per key (same clamp rationale as the server
247/// pager).
248fn stream_virtual(
249    spec: &ViewSpec,
250    segs: &ShardSegs,
251    after: Option<&(IndexValue, Vec<u8>)>,
252    limit: usize,
253    all: &mut Vec<(IndexValue, Vec<u8>)>,
254) {
255    let r = resolver(segs);
256    if let Some(order_seg) = r(&spec.order_by) {
257        let cursor = after.map(|(v, k)| kevy_index::Cursor::new(v.clone(), k.clone()));
258        let mut got = 0usize;
259        let mut scan = order_seg.scan(cursor.as_ref(), spec.order);
260        while let Some((v, k)) = scan.next_entry() {
261            if spec.tree.contains(k, &&r) {
262                all.push((v.clone(), k.to_vec()));
263                got += 1;
264                if got == limit {
265                    break;
266                }
267            }
268        }
269    }
270}
271
272fn rebuild(vs: &mut ViewState, segs: &ShardSegs) {
273    let spec = vs.spec.clone();
274    let Some(mat) = &mut vs.mat else {
275        vs.needs_rebuild = false;
276        return;
277    };
278    mat.clear();
279    let mut rows = eval_shard(&spec, segs);
280    rows.sort();
281    if let ViewMode::Materialized { top_k } = spec.mode
282        && top_k > 0
283    {
284        if spec.order == SortOrder::Desc {
285            rows.reverse();
286        }
287        rows.truncate((top_k + top_k / 4) as usize);
288    }
289    for (v, k) in rows {
290        mat.apply(&k, Membership::Member(Some(v)));
291    }
292    vs.needs_rebuild = false;
293}
294
295/// Reconcile with the catalog (under the shard lock).
296pub(crate) fn sync_views(reg: &ViewReg, sv: &mut ShardViews, segs: &mut ShardSegs) {
297    let g = reg.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
298    let (ver, cat) = &*g;
299    // a view reads its indexes by key: exactly those keep a key directory
300    if sv.dirs_at != (*ver, segs.version) {
301        let mut read: Vec<Vec<u8>> = Vec::new();
302        for v in cat.iter() {
303            read.push(v.order_by.clone());
304            v.tree.each_leaf(&mut |l| read.push(l.index.clone()));
305        }
306        for (spec, seg) in &mut segs.segs {
307            seg.set_key_dir(read.iter().any(|n| n.as_slice() == spec.name()));
308        }
309        sv.dirs_at = (*ver, segs.version);
310    }
311    let segs = &*segs;
312    if sv.version == *ver {
313        return;
314    }
315    sv.mark_stats_dirty();
316    let mut next = Vec::new();
317    for spec in cat.iter() {
318        match sv.views.iter().position(|v| v.spec == *spec) {
319            Some(i) => next.push(sv.views.swap_remove(i)),
320            None => {
321                // only a materialized view keeps a set; every other mode reads at query time
322                let mat = match spec.mode {
323                    ViewMode::Materialized { top_k } => {
324                        Some(MaterializedSet::new(top_k, spec.order))
325                    }
326                    _ => None,
327                };
328                let mut vs = ViewState { spec: spec.clone(), needs_rebuild: mat.is_some(), mat };
329                if vs.needs_rebuild {
330                    rebuild(&mut vs, segs);
331                }
332                next.push(vs);
333            }
334        }
335    }
336    sv.views = next;
337    sv.version = *ver;
338}
339
340/// Write hook — call AFTER `ops_index::on_commit` (same shard lock).
341pub(crate) fn on_commit(reg: &ViewReg, sv: &mut ShardViews, segs: &mut ShardSegs, parts: &[&[u8]]) {
342    {
343        let g = reg.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
344        if g.1.is_empty() {
345            return;
346        }
347    }
348    sync_views(reg, sv, segs);
349    let segs = &*segs;
350    let verb = parts.first().copied().unwrap_or(b"");
351    if verb.eq_ignore_ascii_case(b"FLUSHALL") || verb.eq_ignore_ascii_case(b"FLUSHDB") {
352        for vs in &mut sv.views {
353            if let Some(m) = &mut vs.mat {
354                m.clear();
355            }
356        }
357        sv.mark_stats_dirty();
358        return;
359    }
360    // Same exact written-key walk as the index hook.
361    let mut touched = false;
362    let views = &mut sv.views;
363    crate::ops_index::each_written_key_pub(verb, parts, |key| {
364        for vs in &mut *views {
365            let Some(mat) = &mut vs.mat else { continue };
366            touched = true;
367            let r = resolver(segs);
368            let membership = if vs.spec.tree.contains(key, &&r) {
369                Membership::Member(r(&vs.spec.order_by).and_then(|s| s.key_dir()?.get(key)))
370            } else {
371                Membership::NonMember
372            };
373            if mat.apply(key, membership) {
374                vs.needs_rebuild = true;
375            }
376        }
377    });
378    if touched {
379        sv.mark_stats_dirty();
380    }
381}
kevy_embedded/ops_view.rs

kevy_embedded/
ops_view.rs
