kevy_embedded/
ops_view.rs

1//! v2.6 — 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 std::io;
11use std::sync::RwLock;
12
13use kevy_index::{
14    IndexValue, MaterializedSet, Tree, ViewCatalog, ViewMode, ViewSpec, eval_tree, key_in_tree,
15};
16
17use crate::ops_index::ShardSegs;
18use crate::store::{Store, lock_write};
19
20/// Store-level registry.
21#[derive(Default)]
22pub(crate) struct ViewReg {
23    pub(crate) catalog: RwLock<(u64, ViewCatalog)>,
24}
25
26/// One shard's view states (inside `Inner`, guarded by the shard lock).
27#[derive(Default)]
28pub(crate) struct ShardViews {
29    pub(crate) version: u64,
30    pub(crate) views: Vec<ViewState>,
31}
32
33pub(crate) struct ViewState {
34    spec: ViewSpec,
35    mat: Option<MaterializedSet>,
36    needs_rebuild: bool,
37}
38
39/// One page of view members plus the resume cursor.
40pub type ViewPage = (Vec<(Vec<u8>, IndexValue)>, Option<(IndexValue, Vec<u8>)>);
41
42const SIDECAR: &str = "view-catalog.meta";
43
44impl Store {
45    /// Declare a view (typed tree; `via` is not supported embedded —
46    /// read fields in-process). Builds synchronously.
47    pub fn view_create(
48        &self,
49        name: &[u8],
50        tree: Tree,
51        order_by: &[u8],
52        desc: bool,
53        mode: ViewMode,
54    ) -> io::Result<()> {
55        self.check_view_refs(&tree, order_by)?;
56        let spec = ViewSpec {
57            name: name.to_vec(),
58            tree,
59            order_by: order_by.to_vec(),
60            desc,
61            mode,
62            via: None,
63        };
64        {
65            let mut g = self
66                .views
67                .catalog
68                .write()
69                .unwrap_or_else(std::sync::PoisonError::into_inner);
70            let (ver, cat) = &mut *g;
71            cat.create(spec)
72                .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
73            *ver += 1;
74        }
75        self.persist_view_sidecar();
76        for shard in self.shards.iter() {
77            let mut g = lock_write(shard);
78            let inner = &mut *g;
79            crate::ops_index::sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
80            sync_views(&self.views, &mut inner.view_segs, &inner.idx_segs);
81        }
82        Ok(())
83    }
84
85    /// Every index a view references (its leaves + ORDER BY) must
86    /// already be declared.
87    fn check_view_refs(&self, tree: &Tree, order_by: &[u8]) -> io::Result<()> {
88        let mut names: Vec<Vec<u8>> = vec![order_by.to_vec()];
89        tree.each_leaf(&mut |l| names.push(l.index.clone()));
90        let g = self
91            .indexes
92            .catalog
93            .read()
94            .unwrap_or_else(std::sync::PoisonError::into_inner);
95        for n in &names {
96            if g.1.get(n).is_none() {
97                return Err(io::Error::new(
98                    io::ErrorKind::InvalidInput,
99                    "view references unknown index",
100                ));
101            }
102        }
103        Ok(())
104    }
105
106    /// Drop a view; `false` if absent.
107    pub fn view_drop(&self, name: &[u8]) -> bool {
108        let hit = {
109            let mut g = self
110                .views
111                .catalog
112                .write()
113                .unwrap_or_else(std::sync::PoisonError::into_inner);
114            let (ver, cat) = &mut *g;
115            let hit = cat.drop_view(name);
116            if hit {
117                *ver += 1;
118            }
119            hit
120        };
121        if hit {
122            self.persist_view_sidecar();
123        }
124        hit
125    }
126
127    /// Ordered page across shards (`after` resumes exclusively; DESC
128    /// views page from the large end).
129    pub fn view_query(
130        &self,
131        name: &[u8],
132        after: Option<&(IndexValue, Vec<u8>)>,
133        limit: usize,
134    ) -> io::Result<ViewPage> {
135        let limit = limit.clamp(1, 100_000);
136        let mut desc = false;
137        let mut all: Vec<(IndexValue, Vec<u8>)> = Vec::new();
138        let mut found = false;
139        for shard in self.shards.iter() {
140            let mut g = lock_write(shard);
141            let inner = &mut *g;
142            crate::ops_index::sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
143            sync_views(&self.views, &mut inner.view_segs, &inner.idx_segs);
144            let Some(vs) = inner.view_segs.views.iter_mut().find(|v| v.spec.name == name) else {
145                continue;
146            };
147            found = true;
148            desc = vs.spec.desc;
149            if vs.needs_rebuild {
150                rebuild(vs, &inner.idx_segs);
151            }
152            match &vs.mat {
153                Some(m) => all.extend(m.page(after, limit, vs.spec.desc)),
154                None => stream_virtual(&vs.spec, &inner.idx_segs, after, limit, &mut all),
155            }
156        }
157        if !found {
158            return Err(io::Error::new(io::ErrorKind::NotFound, "no such view"));
159        }
160        all.sort();
161        if desc {
162            all.reverse();
163        }
164        all.truncate(limit);
165        let next = if all.len() == limit { all.last().cloned() } else { None };
166        Ok((all.into_iter().map(|(v, k)| (k, v)).collect(), next))
167    }
168
169    /// Declared views (name, mode, leaves).
170    pub fn view_list(&self) -> Vec<(Vec<u8>, ViewMode, usize)> {
171        let g = self
172            .views
173            .catalog
174            .read()
175            .unwrap_or_else(std::sync::PoisonError::into_inner);
176        g.1.iter()
177            .map(|s| (s.name.clone(), s.mode, s.tree.leaves()))
178            .collect()
179    }
180
181    /// Summed member count across shards.
182    pub fn view_count(&self, name: &[u8]) -> io::Result<u64> {
183        Ok(self.view_query(name, None, 100_000)?.0.len() as u64)
184    }
185
186    fn persist_view_sidecar(&self) {
187        let Some(dir) = &self.config.data_dir else { return };
188        let g = self
189            .views
190            .catalog
191            .read()
192            .unwrap_or_else(std::sync::PoisonError::into_inner);
193        let tmp = dir.join("view-catalog.meta.tmp");
194        if std::fs::write(&tmp, g.1.to_sidecar()).is_ok() {
195            let _ = std::fs::rename(&tmp, dir.join(SIDECAR));
196        }
197    }
198
199    /// Boot half — load the persisted view catalog.
200    pub(crate) fn view_boot(&self) {
201        let Some(dir) = &self.config.data_dir else { return };
202        if let Ok(text) = std::fs::read_to_string(dir.join(SIDECAR))
203            && let Some(cat) = ViewCatalog::from_sidecar(&text)
204            && !cat.is_empty()
205        {
206            let mut g = self
207                .views
208                .catalog
209                .write()
210                .unwrap_or_else(std::sync::PoisonError::into_inner);
211            *g = (g.0 + 1, cat);
212        }
213    }
214}
215
216fn resolver<'a>(segs: &'a ShardSegs) -> impl Fn(&[u8]) -> Option<&'a kevy_index::Segment> {
217    move |name: &[u8]| segs.segs.iter().find(|(s, _)| s.name == name).map(|(_, seg)| seg)
218}
219
220fn eval_shard(spec: &ViewSpec, segs: &ShardSegs) -> Vec<(IndexValue, Vec<u8>)> {
221    let r = resolver(segs);
222    let members = eval_tree(&spec.tree, &&r);
223    members
224        .into_iter()
225        .filter_map(|k| r(&spec.order_by).and_then(|s| s.verify_entry(&k)).map(|v| (v.clone(), k)))
226        .collect()
227}
228
229/// Virtual-mode page: order-driven streaming over the ORDER BY index,
230/// probing the tree per key (same clamp rationale as the server
231/// pager).
232fn stream_virtual(
233    spec: &ViewSpec,
234    segs: &ShardSegs,
235    after: Option<&(IndexValue, Vec<u8>)>,
236    limit: usize,
237    all: &mut Vec<(IndexValue, Vec<u8>)>,
238) {
239    let r = resolver(segs);
240    if let Some(order_seg) = r(&spec.order_by) {
241        let cursor = after.map(|(v, k)| kevy_index::Cursor { value: v.clone(), key: k.clone() });
242        let mut got = 0usize;
243        for (v, k) in order_seg.scan(cursor.as_ref(), spec.desc) {
244            if key_in_tree(&spec.tree, k, &&r) {
245                all.push((v.clone(), k.to_vec()));
246                got += 1;
247                if got == limit {
248                    break;
249                }
250            }
251        }
252    }
253}
254
255fn rebuild(vs: &mut ViewState, segs: &ShardSegs) {
256    let spec = vs.spec.clone();
257    let Some(mat) = &mut vs.mat else {
258        vs.needs_rebuild = false;
259        return;
260    };
261    mat.clear();
262    let mut rows = eval_shard(&spec, segs);
263    rows.sort();
264    if let ViewMode::Materialized { top_k } = spec.mode
265        && top_k > 0
266    {
267        if spec.desc {
268            rows.reverse();
269        }
270        rows.truncate((top_k + top_k / 4) as usize);
271    }
272    for (v, k) in rows {
273        mat.apply(&k, true, Some(v));
274    }
275    vs.needs_rebuild = false;
276}
277
278/// Reconcile with the catalog (under the shard lock).
279pub(crate) fn sync_views(reg: &ViewReg, sv: &mut ShardViews, segs: &ShardSegs) {
280    let g = reg
281        .catalog
282        .read()
283        .unwrap_or_else(std::sync::PoisonError::into_inner);
284    let (ver, cat) = &*g;
285    if sv.version == *ver {
286        return;
287    }
288    let mut next = Vec::new();
289    for spec in cat.iter() {
290        match sv.views.iter().position(|v| v.spec == *spec) {
291            Some(i) => next.push(sv.views.swap_remove(i)),
292            None => {
293                let mat = match spec.mode {
294                    ViewMode::Virtual => None,
295                    ViewMode::Materialized { top_k } => Some(MaterializedSet::new(top_k, spec.desc)),
296                };
297                let mut vs = ViewState { spec: spec.clone(), needs_rebuild: mat.is_some(), mat };
298                if vs.needs_rebuild {
299                    rebuild(&mut vs, segs);
300                }
301                next.push(vs);
302            }
303        }
304    }
305    sv.views = next;
306    sv.version = *ver;
307}
308
309/// Write hook — call AFTER `ops_index::on_commit` (same shard lock).
310pub(crate) fn on_commit(
311    reg: &ViewReg,
312    sv: &mut ShardViews,
313    segs: &ShardSegs,
314    parts: &[&[u8]],
315) {
316    {
317        let g = reg
318            .catalog
319            .read()
320            .unwrap_or_else(std::sync::PoisonError::into_inner);
321        if g.1.is_empty() {
322            return;
323        }
324    }
325    sync_views(reg, sv, segs);
326    let verb = parts.first().copied().unwrap_or(b"");
327    if verb.eq_ignore_ascii_case(b"FLUSHALL") || verb.eq_ignore_ascii_case(b"FLUSHDB") {
328        for vs in &mut sv.views {
329            if let Some(m) = &mut vs.mat {
330                m.clear();
331            }
332        }
333        return;
334    }
335    // Same exact written-key walk as the index hook.
336    crate::ops_index::each_written_key_pub(verb, parts, |key| {
337        for vs in &mut sv.views {
338            let Some(mat) = &mut vs.mat else { continue };
339            let r = resolver(segs);
340            let member = key_in_tree(&vs.spec.tree, key, &&r);
341            let order = r(&vs.spec.order_by).and_then(|s| s.verify_entry(key)).cloned();
342            if mat.apply(key, member, order) {
343                vs.needs_rebuild = true;
344            }
345        }
346    });
347}
kevy_embedded/ops_view.rs

kevy_embedded/
ops_view.rs
