1use std::collections::{BTreeMap, HashSet};
17use std::sync::{Arc, OnceLock};
18
19use core_query::cypher::{execute, is_write_tokens, lex, parse, plan, Params};
20use core_query::{expand, neighborhood, Dir, GraphView, ResultSet};
21use core_storage::fulltext::FulltextIndex;
22use core_storage::v8::seam::{ColumnsView, TopologyView};
23use core_storage::v8::MappedBase;
24use core_storage::wal::WalRecord;
25use core_storage::{
26 ColumnStore, Direction, EdgeProps, EdgePropsView, GraphError, IdMap, Interner, Result,
27 Topology, Value,
28};
29
30use crate::db::{EdgeInfo, NodeInfo};
31use crate::mask::{NodeMask, RoleMaskCache};
32use crate::roles::RoleDef;
33
34pub const FOLD_EVERY_K: usize = 64;
37
38pub struct CommitDelta {
40 pub records: Vec<WalRecord>,
42 pub derived_inserts: Vec<(u32, u32, u32)>,
44 pub derived_deletes: Vec<(u32, u32, u32)>,
46}
47
48#[derive(Clone)]
52pub struct FrozenOverlay {
53 pub ids: Arc<IdMap>,
54 pub syms: Arc<Interner>,
55 pub topo: Arc<Topology>,
56 pub props: Arc<ColumnStore>,
57 pub labels: Arc<Vec<u32>>,
58 pub edge_props: Arc<EdgeProps>,
59 pub roles: Option<Arc<Vec<RoleDef>>>,
60 pub fulltext: Arc<FulltextIndex>,
61}
62
63pub struct ReaderSnapshot {
72 pub frozen: Arc<FrozenOverlay>,
74 pub base: Option<Arc<MappedBase>>,
76 pub deltas: Vec<Arc<CommitDelta>>,
78 pub version: u64,
82 role_masks: Arc<RoleMaskCache>,
84 cache: OnceLock<std::result::Result<FrozenOverlay, String>>,
92}
93
94fn build_tv<'a>(topo: &'a Topology, base: &'a Option<Arc<MappedBase>>) -> TopologyView<'a> {
97 match base {
98 None => TopologyView::owned(topo),
99 Some(b) => {
100 let csr = b.topology().expect("base CSR CRC already verified at open");
101 TopologyView::with_base(topo, csr)
102 }
103 }
104}
105
106fn build_cv<'a>(props: &'a ColumnStore, base: &'a Option<Arc<MappedBase>>) -> ColumnsView<'a> {
107 match base {
108 None => ColumnsView::owned(props),
109 Some(b) => {
110 let cols = b
111 .columns()
112 .expect("base columns CRC already verified at open");
113 let strings = b
114 .string_table()
115 .transpose()
116 .expect("base strings CRC already verified at open");
117 ColumnsView::with_base_cached(props, cols, b.mixed_cache()).with_shared_strings(strings)
118 }
119 }
120}
121
122fn build_epv<'a>(
123 edge_props: &'a EdgeProps,
124 base: &'a Option<Arc<MappedBase>>,
125) -> EdgePropsView<'a> {
126 match base {
127 None => EdgePropsView::owned(edge_props),
128 Some(b) => {
129 let archived = b
130 .edge_props_section()
131 .expect("base edge_props CRC already verified at open");
132 EdgePropsView::with_base(edge_props, archived)
133 }
134 }
135}
136
137fn make_view<'a>(
138 state: &'a FrozenOverlay,
139 base: &'a Option<Arc<MappedBase>>,
140 mask: Option<&'a core_query::visible::VisibleSet>,
141) -> GraphView<'a> {
142 GraphView {
143 ids: &state.ids,
144 syms: &state.syms,
145 labels: &state.labels,
146 props: build_cv(&state.props, base),
147 topo: build_tv(&state.topo, base),
148 edge_props: build_epv(&state.edge_props, base),
149 mask,
150 prop_index: None,
153 }
154}
155
156#[allow(clippy::too_many_arguments)]
165fn apply_one(
166 ids: &mut IdMap,
167 syms: &mut Interner,
168 topo: &mut Topology,
169 props: &mut ColumnStore,
170 edge_props: &mut EdgeProps,
171 labels: &mut Vec<u32>,
172 fulltext: &mut FulltextIndex,
173 base: &Option<Arc<MappedBase>>,
174 rec: &WalRecord,
175) -> Result<()> {
176 match rec {
177 WalRecord::Intern { id, text } => {
178 let got = syms.intern(text);
179 if got != *id {
180 return Err(GraphError::Corrupt {
181 detail: format!(
182 "mvcc delta intern mismatch for {text:?}: expected {id}, got {got}"
183 ),
184 });
185 }
186 }
187
188 WalRecord::InsertNodeId {
189 label,
190 key,
191 props: node_props,
192 } => {
193 let node_id = ids.try_insert(key)?;
194 if labels.len() <= node_id as usize {
195 labels.resize(node_id as usize + 1, u32::MAX);
196 }
197 labels[node_id as usize] = *label;
198 let label_str = syms
199 .resolve(*label)
200 .ok_or_else(|| GraphError::Corrupt {
201 detail: format!("mvcc delta: unknown label sym {label}"),
202 })?
203 .to_string();
204 for (field_sym, value) in node_props {
205 let field = syms
206 .resolve(*field_sym)
207 .ok_or_else(|| GraphError::Corrupt {
208 detail: format!("mvcc delta: unknown field sym {field_sym}"),
209 })?
210 .to_string();
211 props.set(node_id, &field, value.clone());
212 if fulltext.is_enabled(&label_str, &field) {
213 fulltext.add_tokens(node_id, &field, value);
214 }
215 }
216 }
217
218 WalRecord::InsertNode {
219 label,
220 key,
221 props: node_props,
222 } => {
223 let label_sym = syms.intern(label);
224 let node_id = ids.try_insert(key)?;
225 if labels.len() <= node_id as usize {
226 labels.resize(node_id as usize + 1, u32::MAX);
227 }
228 labels[node_id as usize] = label_sym;
229 for (field, value) in node_props {
230 props.set(node_id, field, value.clone());
231 if fulltext.is_enabled(label, field) {
232 fulltext.add_tokens(node_id, field, value);
233 }
234 }
235 }
236
237 WalRecord::SetPropId { id, field, value } => {
238 if let Some(field_str) = syms.resolve(*field).map(str::to_string) {
239 props.set(*id, &field_str, value.clone());
240 if let Some(&label_sym) = labels.get(*id as usize) {
241 if let Some(label_str) = syms.resolve(label_sym) {
242 if fulltext.is_enabled(label_str, &field_str) {
243 fulltext.add_tokens(*id, &field_str, value);
244 }
245 }
246 }
247 }
248 }
249
250 WalRecord::SetProp { key, field, value } => {
251 if let Some(node_id) = ids.get(key) {
252 props.set(node_id, field, value.clone());
253 if let Some(&label_sym) = labels.get(node_id as usize) {
254 if let Some(label_str) = syms.resolve(label_sym) {
255 if fulltext.is_enabled(label_str, field) {
256 fulltext.add_tokens(node_id, field, value);
257 }
258 }
259 }
260 }
261 }
262
263 WalRecord::RemoveProp { key, field } => {
264 if let Some(node_id) = ids.get(key) {
265 props.remove(node_id, field);
266 props.record_prop_tombstone(node_id, field);
271 fulltext.remove_node_field(node_id, field);
272 }
273 }
274
275 WalRecord::DeleteNode { key } => {
276 if let Some(node_id) = ids.delete(key) {
277 props.remove_all(node_id);
278 fulltext.remove_node(node_id);
279 if let Some(slot) = labels.get_mut(node_id as usize) {
281 *slot = u32::MAX;
282 }
283 let mut doomed = Vec::new();
289 {
290 let tv = match base {
291 None => TopologyView::owned(topo),
292 Some(b) => match b.topology() {
293 Ok(archived) => TopologyView::with_base(topo, archived),
294 Err(_) => TopologyView::owned(topo),
295 },
296 };
297 let etypes: Vec<u32> = tv.etypes().collect();
298 for et in &etypes {
299 for &dst in tv.neighbors(*et, Direction::Out, node_id).as_ref() {
300 doomed.push((*et, node_id, dst));
301 }
302 for &src in tv.neighbors(*et, Direction::In, node_id).as_ref() {
303 if src != node_id {
304 doomed.push((*et, src, node_id));
305 }
306 }
307 }
308 }
309 for (et, s, d) in doomed {
310 topo.remove_edge(et, s, d);
311 edge_props.remove_edge(et, s, d);
312 }
313 }
314 }
315
316 WalRecord::InsertEdgeId { etype, src, dst } => {
317 topo.add_edge(*etype, *src, *dst);
318 }
319
320 WalRecord::InsertEdge {
321 edge_type,
322 src_key,
323 dst_key,
324 } => {
325 let etype = syms.intern(edge_type);
326 if let (Some(src), Some(dst)) = (ids.get(src_key), ids.get(dst_key)) {
327 topo.add_edge(etype, src, dst);
328 }
329 }
330
331 WalRecord::DeleteEdge {
332 edge_type,
333 src_key,
334 dst_key,
335 } => {
336 if let Some(etype) = syms.get(edge_type) {
337 if let (Some(src), Some(dst)) = (ids.get(src_key), ids.get(dst_key)) {
338 topo.remove_edge(etype, src, dst);
339 }
340 }
341 }
342
343 WalRecord::EnableFulltext { label, field } => {
344 fulltext.enable(label, field);
345 }
346
347 WalRecord::DisableFulltext { label, field } => {
348 fulltext.disable(label, field);
349 }
350
351 WalRecord::Batch(inner) => {
352 for r in inner {
353 apply_one(ids, syms, topo, props, edge_props, labels, fulltext, base, r)?;
354 }
355 }
356
357 WalRecord::CreateRule { .. }
360 | WalRecord::DeleteRule { .. }
361 | WalRecord::RebuildRule { .. }
362 | WalRecord::CreateView { .. }
363 | WalRecord::DeleteView { .. }
364 | WalRecord::EnableIndex { .. }
368 | WalRecord::DisableIndex { .. }
369 | WalRecord::SetEdgeCount { count: 0, .. }
373 | WalRecord::DerivedEdgeAdded { .. }
377 | WalRecord::DerivedEdgeRetracted { .. } => {}
378
379 WalRecord::SetEdgeCount {
382 etype,
383 src,
384 dst,
385 count,
386 } => {
387 edge_props.set(
388 *etype,
389 *src,
390 *dst,
391 crate::db::EDGE_COUNT_PROP,
392 core_storage::Value::Int(*count as i64),
393 );
394 }
395
396 WalRecord::RenameNode { old_key, new_key } => {
397 if ids.get(old_key).is_some() {
400 ids.rename(old_key, new_key).map_err(|_| GraphError::Corrupt {
401 detail: format!("mvcc delta RenameNode {old_key}→{new_key} failed"),
402 })?;
403 }
404 }
405 }
406 Ok(())
407}
408
409fn mask_for_role_from(
418 state: &FrozenOverlay,
419 base: &Option<Arc<MappedBase>>,
420 role: &str,
421) -> Result<NodeMask> {
422 let roles = state.roles.as_ref().ok_or_else(|| GraphError::Corrupt {
423 detail: "roles.json was corrupt at open; fix the file and re-open".into(),
424 })?;
425 let def = roles
426 .iter()
427 .find(|r| r.name == role)
428 .ok_or_else(|| GraphError::KeyNotFound {
429 key: format!("role:{role}"),
430 })?;
431 let mut visible = HashSet::new();
432 for key in &def.keys {
434 if let Some(id) = state.ids.get(key) {
435 visible.insert(id);
436 }
437 }
438 let props = def
439 .visible_where
440 .as_ref()
441 .map(|_| build_cv(&state.props, base));
442 for label_name in &def.labels {
443 if let Some(sym) = state.syms.get(label_name) {
444 for (i, &s) in state.labels.iter().enumerate() {
445 if s != sym {
446 continue;
447 }
448 let id = i as u32;
449 match (&def.visible_where, &props) {
450 (Some(pred), Some(view)) => {
451 let value = view.get(id, &pred.field).map(|vr| vr.into_value());
452 if pred.holds(value.as_ref()) {
453 visible.insert(id);
454 }
455 }
456 _ => {
457 visible.insert(id);
458 }
459 }
460 }
461 }
462 }
463 if def.namespaces.is_some() {
469 let cv = build_cv(&state.props, base);
470 visible.retain(|&id| {
471 let value = cv.get(id, core_storage::NS_PROP).map(|vr| vr.into_value());
472 def.sees_namespace(core_storage::namespace_of_value(value.as_ref()))
473 });
474 }
475 Ok(NodeMask::from_ids(visible))
476}
477
478impl ReaderSnapshot {
479 fn materialize(&self) -> Result<FrozenOverlay> {
484 let mut w = (*self.frozen).clone();
494 for delta in &self.deltas {
495 for rec in &delta.records {
496 apply_one(
497 Arc::make_mut(&mut w.ids),
498 Arc::make_mut(&mut w.syms),
499 Arc::make_mut(&mut w.topo),
500 Arc::make_mut(&mut w.props),
501 Arc::make_mut(&mut w.edge_props),
502 Arc::make_mut(&mut w.labels),
503 Arc::make_mut(&mut w.fulltext),
504 &self.base,
505 rec,
506 )?;
507 }
508 for &(etype, src, dst) in &delta.derived_inserts {
509 Arc::make_mut(&mut w.topo).add_edge(etype, src, dst);
510 }
511 for &(etype, src, dst) in &delta.derived_deletes {
512 Arc::make_mut(&mut w.topo).remove_edge(etype, src, dst);
513 }
514 }
515 if !self.deltas.is_empty() {
516 let cv = build_cv(&w.props, &self.base);
519 let (ids, labels, syms) = (w.ids.clone(), w.labels.clone(), w.syms.clone());
520 Arc::make_mut(&mut w.fulltext).rebuild_all(&ids, &labels, &syms, cv);
521 }
522 Ok(w)
523 }
524
525 pub(crate) fn new(
530 frozen: Arc<FrozenOverlay>,
531 base: Option<Arc<MappedBase>>,
532 deltas: Vec<Arc<CommitDelta>>,
533 version: u64,
534 role_masks: Arc<RoleMaskCache>,
535 ) -> Self {
536 Self {
537 frozen,
538 base,
539 deltas,
540 version,
541 role_masks,
542 cache: OnceLock::new(),
543 }
544 }
545
546 fn effective(&self) -> Result<&FrozenOverlay> {
556 if self.deltas.is_empty() {
557 return Ok(&self.frozen);
558 }
559 let cached = self
560 .cache
561 .get_or_init(|| self.materialize().map_err(|e| e.to_string()));
562 cached
563 .as_ref()
564 .map_err(|e| GraphError::Corrupt { detail: e.clone() })
565 }
566
567 pub fn mask_for_role(&self, role: &str) -> Result<NodeMask> {
579 self.role_masks
580 .get_or_build(role, self.version, || {
581 mask_for_role_from(self.effective()?, &self.base, role)
582 })
583 .map(|m| (*m).clone())
584 }
585
586 pub fn mask_for_namespace(&self, namespace: &str) -> Result<NodeMask> {
593 let state = self.effective()?;
594 let cv = build_cv(&state.props, &self.base);
595 let mut visible = HashSet::new();
596 for (i, &sym) in state.labels.iter().enumerate() {
597 if sym == u32::MAX {
598 continue; }
600 let id = i as u32;
601 let value = cv.get(id, core_storage::NS_PROP).map(|vr| vr.into_value());
602 if core_storage::namespace_of_value(value.as_ref()) == namespace {
603 visible.insert(id);
604 }
605 }
606 Ok(NodeMask::from_ids(visible))
607 }
608
609 pub fn resolve_key(&self, key: &str) -> Option<u32> {
614 self.effective().ok()?.ids.get(key)
615 }
616
617 pub fn query(&self, cypher: &str, params: &BTreeMap<String, Value>) -> Result<ResultSet> {
619 let tokens = lex(cypher).map_err(|e| GraphError::QueryError {
620 detail: format!("lex: {e}"),
621 })?;
622 let ast = parse(&tokens).map_err(|e| GraphError::QueryError {
623 detail: format!("parse: {e}"),
624 })?;
625 let ops = plan(&ast).map_err(|e| GraphError::QueryError {
626 detail: format!("plan: {e}"),
627 })?;
628 let state = self.effective()?;
629 let view = make_view(state, &self.base, None);
630 execute(&view, &ops, &Params(params)).map_err(|e| GraphError::QueryError {
631 detail: format!("execute: {e}"),
632 })
633 }
634
635 pub fn query_masked(
639 &self,
640 cypher: &str,
641 params: &BTreeMap<String, Value>,
642 mask: &NodeMask,
643 ) -> Result<ResultSet> {
644 let tokens = lex(cypher).map_err(|e| GraphError::QueryError {
645 detail: format!("lex: {e}"),
646 })?;
647 if is_write_tokens(&tokens) {
648 return Err(GraphError::QueryError {
649 detail: "masked queries are read-only".into(),
650 });
651 }
652 let ast = parse(&tokens).map_err(|e| GraphError::QueryError {
653 detail: format!("parse: {e}"),
654 })?;
655 let ops = plan(&ast).map_err(|e| GraphError::QueryError {
656 detail: format!("plan: {e}"),
657 })?;
658 let state = self.effective()?;
659 let view = make_view(state, &self.base, Some(&mask.visible));
660 execute(&view, &ops, &Params(params)).map_err(|e| GraphError::QueryError {
661 detail: format!("execute: {e}"),
662 })
663 }
664
665 pub fn node_info(&self, key: &str) -> Option<NodeInfo> {
667 node_info_from(key, self.effective().ok()?, &self.base)
668 }
669
670 pub fn node_edges(&self, key: &str) -> Result<Vec<EdgeInfo>> {
675 node_edges_from(key, self.effective()?, &self.base)
676 }
677
678 pub fn neighborhood_masked(
683 &self,
684 key: &str,
685 depth: u32,
686 edge_types: Option<&[&str]>,
687 dir: Dir,
688 mask: &NodeMask,
689 ) -> Option<ResultSet> {
690 neighborhood_masked_from(
691 key,
692 self.effective().ok()?,
693 &self.base,
694 depth,
695 edge_types,
696 dir,
697 mask,
698 )
699 }
700
701 pub fn node_edges_scoped(&self, key: &str, mask: &NodeMask) -> Result<Vec<EdgeInfo>> {
715 let state = self.effective()?;
716 if !state.ids.get(key).is_some_and(|id| mask.contains_id(id)) {
717 return Err(GraphError::KeyNotFound { key: key.into() });
718 }
719 let edges = node_edges_from(key, state, &self.base)?;
720 Ok(edges
721 .into_iter()
722 .filter(|e| {
723 let other = if e.src_key == key {
724 &e.dst_key
725 } else {
726 &e.src_key
727 };
728 state.ids.get(other).is_some_and(|id| mask.contains_id(id))
729 })
730 .collect())
731 }
732
733 pub fn neighborhood_scoped(
743 &self,
744 key: &str,
745 depth: u32,
746 edge_types: Option<&[&str]>,
747 dir: Dir,
748 mask: &NodeMask,
749 ) -> Result<ResultSet> {
750 let state = self.effective()?;
751 if !state.ids.get(key).is_some_and(|id| mask.contains_id(id)) {
752 return Err(GraphError::KeyNotFound { key: key.into() });
753 }
754 neighborhood_masked_from(key, state, &self.base, depth, edge_types, dir, mask)
755 .ok_or_else(|| GraphError::KeyNotFound { key: key.into() })
756 }
757}
758
759fn node_info_from(
762 key: &str,
763 state: &FrozenOverlay,
764 base: &Option<Arc<MappedBase>>,
765) -> Option<NodeInfo> {
766 let id = state.ids.get(key)?;
767 let label_sym = *state.labels.get(id as usize)?;
768 if label_sym == u32::MAX {
769 return None;
770 }
771 let label = state.syms.resolve(label_sym)?.to_string();
772 let cv = build_cv(&state.props, base);
773 let mut props = BTreeMap::new();
774 for field in cv.field_names() {
775 if let Some(vr) = cv.get(id, &field) {
776 props.insert(field, vr.into_value());
777 }
778 }
779 Some(NodeInfo {
780 key: key.to_string(),
781 label,
782 props,
783 })
784}
785
786fn node_edges_from(
787 key: &str,
788 state: &FrozenOverlay,
789 base: &Option<Arc<MappedBase>>,
790) -> Result<Vec<EdgeInfo>> {
791 let id = state
792 .ids
793 .get(key)
794 .ok_or_else(|| GraphError::KeyNotFound { key: key.into() })?;
795 let tv = build_tv(&state.topo, base);
796 let mut edges = Vec::new();
797 for etype in tv.etypes() {
798 let edge_type = state
799 .syms
800 .resolve(etype)
801 .ok_or_else(|| GraphError::Corrupt {
802 detail: format!("reader: topology etype {etype} not in interner"),
803 })?
804 .to_string();
805 for dir in [Direction::Out, Direction::In] {
806 for &nbr in tv.neighbors(etype, dir, id).as_ref() {
807 let (src_key, dst_key) = match dir {
808 Direction::Out => (
809 key.to_string(),
810 state
811 .ids
812 .key_of(nbr)
813 .ok_or_else(|| GraphError::Corrupt {
814 detail: format!("topology id {nbr} has no key"),
815 })?
816 .to_string(),
817 ),
818 Direction::In => (
819 state
820 .ids
821 .key_of(nbr)
822 .ok_or_else(|| GraphError::Corrupt {
823 detail: format!("topology id {nbr} has no key"),
824 })?
825 .to_string(),
826 key.to_string(),
827 ),
828 };
829 edges.push(EdgeInfo {
830 edge_type: edge_type.clone(),
831 src_key,
832 dst_key,
833 derived: false,
834 });
835 }
836 }
837 }
838 edges.sort_by(|a, b| {
839 a.edge_type
840 .cmp(&b.edge_type)
841 .then(a.src_key.cmp(&b.src_key))
842 .then(a.dst_key.cmp(&b.dst_key))
843 });
844 edges.dedup();
845 Ok(edges)
846}
847
848fn neighborhood_masked_from(
849 key: &str,
850 state: &FrozenOverlay,
851 base: &Option<Arc<MappedBase>>,
852 depth: u32,
853 edge_types: Option<&[&str]>,
854 dir: Dir,
855 mask: &NodeMask,
856) -> Option<ResultSet> {
857 let start_id = state.ids.get(key)?;
858 let view = make_view(state, base, Some(&mask.visible));
859 let resolved: Option<Vec<u32>> = edge_types.map(|names| {
860 names
861 .iter()
862 .filter_map(|name| view.syms.get(name))
863 .collect()
864 });
865 let nb = neighborhood(&view, start_id, depth, resolved.as_deref(), dir);
866 let mut rs = ResultSet::new(vec!["key".into(), "label".into(), "depth".into()]);
867 let mut visited: Vec<(u32, u32)> = Vec::with_capacity(nb.nodes.len() + 1);
869 visited.push((start_id, 0));
870 for (nid, d) in &nb.nodes {
871 let k = view.key_of(*nid);
872 let lbl = view
873 .label_of(*nid)
874 .expect("real nodes always have a label; u32::MAX sentinel cannot occur");
875 rs.push_row(vec![
876 Some(Value::Str(k.to_string())),
877 Some(Value::Str(lbl.to_string())),
878 Some(Value::Int(*d as i64)),
879 ]);
880 visited.push((*nid, *d));
881 }
882 if mask.mode() == crate::mask::MaskMode::Stub {
890 let raw_view = make_view(state, base, None);
891 let mut seen: HashSet<u32> = visited.iter().map(|(id, _)| *id).collect();
892 for (node_id, node_depth) in &visited {
893 if *node_depth >= depth {
894 continue;
895 }
896 for e in expand(&raw_view, *node_id, resolved.as_deref(), dir) {
897 let nbr = if e.src == *node_id { e.dst } else { e.src };
898 if !mask.contains_id(nbr) && seen.insert(nbr) {
899 if let Some(k) = state.ids.key_of(nbr) {
900 rs.push_row(vec![
901 Some(Value::Str(k.to_string())),
902 None,
903 Some(Value::Int((*node_depth + 1) as i64)),
904 ]);
905 }
906 }
907 }
908 }
909 }
910 Some(rs)
911}