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

kevy_embedded/
ops_view.rs
