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 pub view_values: Vec<(u32, String, Option<Value>)>,
52}
53
54#[derive(Clone)]
58pub struct FrozenOverlay {
59 pub ids: Arc<IdMap>,
60 pub syms: Arc<Interner>,
61 pub topo: Arc<Topology>,
62 pub props: Arc<ColumnStore>,
63 pub labels: Arc<Vec<u32>>,
64 pub edge_props: Arc<EdgeProps>,
65 pub roles: Option<Arc<Vec<RoleDef>>>,
66 pub fulltext: Arc<FulltextIndex>,
67}
68
69pub struct ReaderSnapshot {
78 pub frozen: Arc<FrozenOverlay>,
80 pub base: Option<Arc<MappedBase>>,
82 pub deltas: Vec<Arc<CommitDelta>>,
84 pub version: u64,
88 role_masks: Arc<RoleMaskCache>,
90 cache: OnceLock<std::result::Result<FrozenOverlay, String>>,
98}
99
100fn build_tv<'a>(topo: &'a Topology, base: &'a Option<Arc<MappedBase>>) -> TopologyView<'a> {
103 match base {
104 None => TopologyView::owned(topo),
105 Some(b) => {
106 let csr = b.topology().expect("base CSR CRC already verified at open");
107 TopologyView::with_base(topo, csr)
108 }
109 }
110}
111
112fn build_cv<'a>(props: &'a ColumnStore, base: &'a Option<Arc<MappedBase>>) -> ColumnsView<'a> {
113 match base {
114 None => ColumnsView::owned(props),
115 Some(b) => {
116 let cols = b
117 .columns()
118 .expect("base columns CRC already verified at open");
119 let strings = b
120 .string_table()
121 .transpose()
122 .expect("base strings CRC already verified at open");
123 ColumnsView::with_base_cached(props, cols, b.mixed_cache()).with_shared_strings(strings)
124 }
125 }
126}
127
128fn build_epv<'a>(
129 edge_props: &'a EdgeProps,
130 base: &'a Option<Arc<MappedBase>>,
131) -> EdgePropsView<'a> {
132 match base {
133 None => EdgePropsView::owned(edge_props),
134 Some(b) => {
135 let archived = b
136 .edge_props_section()
137 .expect("base edge_props CRC already verified at open");
138 EdgePropsView::with_base(edge_props, archived)
139 }
140 }
141}
142
143fn make_view<'a>(
144 state: &'a FrozenOverlay,
145 base: &'a Option<Arc<MappedBase>>,
146 mask: Option<&'a core_query::visible::VisibleSet>,
147) -> GraphView<'a> {
148 GraphView {
149 ids: &state.ids,
150 syms: &state.syms,
151 labels: &state.labels,
152 props: build_cv(&state.props, base),
153 topo: build_tv(&state.topo, base),
154 edge_props: build_epv(&state.edge_props, base),
155 mask,
156 prop_index: None,
159 }
160}
161
162#[allow(clippy::too_many_arguments)]
171fn apply_one(
172 ids: &mut IdMap,
173 syms: &mut Interner,
174 topo: &mut Topology,
175 props: &mut ColumnStore,
176 edge_props: &mut EdgeProps,
177 labels: &mut Vec<u32>,
178 fulltext: &mut FulltextIndex,
179 base: &Option<Arc<MappedBase>>,
180 rec: &WalRecord,
181) -> Result<()> {
182 match rec {
183 WalRecord::Intern { id, text } => {
184 let got = syms.intern(text);
185 if got != *id {
186 return Err(GraphError::Corrupt {
187 detail: format!(
188 "mvcc delta intern mismatch for {text:?}: expected {id}, got {got}"
189 ),
190 });
191 }
192 }
193
194 WalRecord::InsertNodeId {
195 label,
196 key,
197 props: node_props,
198 } => {
199 let node_id = ids.try_insert(key)?;
200 if labels.len() <= node_id as usize {
201 labels.resize(node_id as usize + 1, u32::MAX);
202 }
203 labels[node_id as usize] = *label;
204 let label_str = syms
205 .resolve(*label)
206 .ok_or_else(|| GraphError::Corrupt {
207 detail: format!("mvcc delta: unknown label sym {label}"),
208 })?
209 .to_string();
210 for (field_sym, value) in node_props {
211 let field = syms
212 .resolve(*field_sym)
213 .ok_or_else(|| GraphError::Corrupt {
214 detail: format!("mvcc delta: unknown field sym {field_sym}"),
215 })?
216 .to_string();
217 props.set(node_id, &field, value.clone());
218 if fulltext.is_enabled(&label_str, &field) {
219 fulltext.add_tokens(node_id, &field, value);
220 }
221 }
222 }
223
224 WalRecord::InsertNode {
225 label,
226 key,
227 props: node_props,
228 } => {
229 let label_sym = syms.intern(label);
230 let node_id = ids.try_insert(key)?;
231 if labels.len() <= node_id as usize {
232 labels.resize(node_id as usize + 1, u32::MAX);
233 }
234 labels[node_id as usize] = label_sym;
235 for (field, value) in node_props {
236 props.set(node_id, field, value.clone());
237 if fulltext.is_enabled(label, field) {
238 fulltext.add_tokens(node_id, field, value);
239 }
240 }
241 }
242
243 WalRecord::SetPropId { id, field, value } => {
244 if let Some(field_str) = syms.resolve(*field).map(str::to_string) {
245 props.set(*id, &field_str, value.clone());
246 if let Some(&label_sym) = labels.get(*id as usize) {
247 if let Some(label_str) = syms.resolve(label_sym) {
248 if fulltext.is_enabled(label_str, &field_str) {
249 fulltext.add_tokens(*id, &field_str, value);
250 }
251 }
252 }
253 }
254 }
255
256 WalRecord::SetProp { key, field, value } => {
257 if let Some(node_id) = ids.get(key) {
258 props.set(node_id, field, value.clone());
259 if let Some(&label_sym) = labels.get(node_id as usize) {
260 if let Some(label_str) = syms.resolve(label_sym) {
261 if fulltext.is_enabled(label_str, field) {
262 fulltext.add_tokens(node_id, field, value);
263 }
264 }
265 }
266 }
267 }
268
269 WalRecord::RemoveProp { key, field } => {
270 if let Some(node_id) = ids.get(key) {
271 props.remove(node_id, field);
272 props.record_prop_tombstone(node_id, field);
277 fulltext.remove_node_field(node_id, field);
278 }
279 }
280
281 WalRecord::DeleteNode { key } => {
282 if let Some(node_id) = ids.delete(key) {
283 props.remove_all(node_id);
284 fulltext.remove_node(node_id);
285 if let Some(slot) = labels.get_mut(node_id as usize) {
287 *slot = u32::MAX;
288 }
289 let mut doomed = Vec::new();
295 {
296 let tv = match base {
297 None => TopologyView::owned(topo),
298 Some(b) => match b.topology() {
299 Ok(archived) => TopologyView::with_base(topo, archived),
300 Err(_) => TopologyView::owned(topo),
301 },
302 };
303 let etypes: Vec<u32> = tv.etypes().collect();
304 for et in &etypes {
305 for &dst in tv.neighbors(*et, Direction::Out, node_id).as_ref() {
306 doomed.push((*et, node_id, dst));
307 }
308 for &src in tv.neighbors(*et, Direction::In, node_id).as_ref() {
309 if src != node_id {
310 doomed.push((*et, src, node_id));
311 }
312 }
313 }
314 }
315 for (et, s, d) in doomed {
316 topo.remove_edge(et, s, d);
317 edge_props.remove_edge(et, s, d);
318 }
319 }
320 }
321
322 WalRecord::InsertEdgeId { etype, src, dst } => {
323 topo.add_edge(*etype, *src, *dst);
324 }
325
326 WalRecord::InsertEdge {
327 edge_type,
328 src_key,
329 dst_key,
330 } => {
331 let etype = syms.intern(edge_type);
332 if let (Some(src), Some(dst)) = (ids.get(src_key), ids.get(dst_key)) {
333 topo.add_edge(etype, src, dst);
334 }
335 }
336
337 WalRecord::DeleteEdge {
338 edge_type,
339 src_key,
340 dst_key,
341 } => {
342 if let Some(etype) = syms.get(edge_type) {
343 if let (Some(src), Some(dst)) = (ids.get(src_key), ids.get(dst_key)) {
344 topo.remove_edge(etype, src, dst);
345 }
346 }
347 }
348
349 WalRecord::EnableFulltext { label, field } => {
350 fulltext.enable(label, field);
351 }
352
353 WalRecord::DisableFulltext { label, field } => {
354 fulltext.disable(label, field);
355 }
356
357 WalRecord::Batch(inner) => {
358 for r in inner {
359 apply_one(ids, syms, topo, props, edge_props, labels, fulltext, base, r)?;
360 }
361 }
362
363 WalRecord::CreateRule { .. }
366 | WalRecord::DeleteRule { .. }
367 | WalRecord::RebuildRule { .. }
368 | WalRecord::CreateView { .. }
369 | WalRecord::DeleteView { .. }
370 | WalRecord::EnableIndex { .. }
374 | WalRecord::DisableIndex { .. }
375 | WalRecord::SetEdgeCount { count: 0, .. }
379 | WalRecord::DerivedEdgeAdded { .. }
383 | WalRecord::DerivedEdgeRetracted { .. } => {}
384
385 WalRecord::SetEdgeCount {
388 etype,
389 src,
390 dst,
391 count,
392 } => {
393 edge_props.set(
394 *etype,
395 *src,
396 *dst,
397 crate::db::EDGE_COUNT_PROP,
398 core_storage::Value::Int(*count as i64),
399 );
400 }
401
402 WalRecord::RenameNode { old_key, new_key } => {
403 if ids.get(old_key).is_some() {
406 ids.rename(old_key, new_key).map_err(|_| GraphError::Corrupt {
407 detail: format!("mvcc delta RenameNode {old_key}→{new_key} failed"),
408 })?;
409 }
410 }
411 }
412 Ok(())
413}
414
415fn mask_for_role_from(
424 state: &FrozenOverlay,
425 base: &Option<Arc<MappedBase>>,
426 role: &str,
427) -> Result<NodeMask> {
428 let roles = state.roles.as_ref().ok_or_else(|| GraphError::Corrupt {
429 detail: "roles.json was corrupt at open; fix the file and re-open".into(),
430 })?;
431 let def = roles
432 .iter()
433 .find(|r| r.name == role)
434 .ok_or_else(|| GraphError::KeyNotFound {
435 key: format!("role:{role}"),
436 })?;
437 let mut visible = HashSet::new();
438 for key in &def.keys {
440 if let Some(id) = state.ids.get(key) {
441 visible.insert(id);
442 }
443 }
444 let props = def
445 .visible_where
446 .as_ref()
447 .map(|_| build_cv(&state.props, base));
448 for label_name in &def.labels {
449 if let Some(sym) = state.syms.get(label_name) {
450 for (i, &s) in state.labels.iter().enumerate() {
451 if s != sym {
452 continue;
453 }
454 let id = i as u32;
455 match (&def.visible_where, &props) {
456 (Some(pred), Some(view)) => {
457 let value = view.get(id, &pred.field).map(|vr| vr.into_value());
458 if pred.holds(value.as_ref()) {
459 visible.insert(id);
460 }
461 }
462 _ => {
463 visible.insert(id);
464 }
465 }
466 }
467 }
468 }
469 if def.namespaces.is_some() {
475 let cv = build_cv(&state.props, base);
476 visible.retain(|&id| {
477 let value = cv.get(id, core_storage::NS_PROP).map(|vr| vr.into_value());
478 def.sees_namespace(core_storage::namespace_of_value(value.as_ref()))
479 });
480 }
481 Ok(NodeMask::from_ids(visible))
482}
483
484impl ReaderSnapshot {
485 fn materialize(&self) -> Result<FrozenOverlay> {
490 let mut w = (*self.frozen).clone();
500 for delta in &self.deltas {
501 for rec in &delta.records {
502 apply_one(
503 Arc::make_mut(&mut w.ids),
504 Arc::make_mut(&mut w.syms),
505 Arc::make_mut(&mut w.topo),
506 Arc::make_mut(&mut w.props),
507 Arc::make_mut(&mut w.edge_props),
508 Arc::make_mut(&mut w.labels),
509 Arc::make_mut(&mut w.fulltext),
510 &self.base,
511 rec,
512 )?;
513 }
514 for &(etype, src, dst) in &delta.derived_inserts {
515 Arc::make_mut(&mut w.topo).add_edge(etype, src, dst);
516 }
517 for &(etype, src, dst) in &delta.derived_deletes {
518 Arc::make_mut(&mut w.topo).remove_edge(etype, src, dst);
519 }
520 for (node, view_prop, value) in &delta.view_values {
521 let props = Arc::make_mut(&mut w.props);
522 match value {
523 Some(v) => props.set(*node, view_prop, v.clone()),
524 None => {
525 props.remove(*node, view_prop);
528 props.record_prop_tombstone(*node, view_prop);
529 }
530 }
531 }
532 }
533 if !self.deltas.is_empty() {
534 let cv = build_cv(&w.props, &self.base);
537 let (ids, labels, syms) = (w.ids.clone(), w.labels.clone(), w.syms.clone());
538 Arc::make_mut(&mut w.fulltext).rebuild_all(&ids, &labels, &syms, cv);
539 }
540 Ok(w)
541 }
542
543 pub(crate) fn new(
548 frozen: Arc<FrozenOverlay>,
549 base: Option<Arc<MappedBase>>,
550 deltas: Vec<Arc<CommitDelta>>,
551 version: u64,
552 role_masks: Arc<RoleMaskCache>,
553 ) -> Self {
554 Self {
555 frozen,
556 base,
557 deltas,
558 version,
559 role_masks,
560 cache: OnceLock::new(),
561 }
562 }
563
564 fn effective(&self) -> Result<&FrozenOverlay> {
574 if self.deltas.is_empty() {
575 return Ok(&self.frozen);
576 }
577 let cached = self
578 .cache
579 .get_or_init(|| self.materialize().map_err(|e| e.to_string()));
580 cached
581 .as_ref()
582 .map_err(|e| GraphError::Corrupt { detail: e.clone() })
583 }
584
585 pub fn mask_for_role(&self, role: &str) -> Result<NodeMask> {
597 self.role_masks
598 .get_or_build(role, self.version, || {
599 mask_for_role_from(self.effective()?, &self.base, role)
600 })
601 .map(|m| (*m).clone())
602 }
603
604 pub fn mask_for_namespace(&self, namespace: &str) -> Result<NodeMask> {
611 let state = self.effective()?;
612 let cv = build_cv(&state.props, &self.base);
613 let mut visible = HashSet::new();
614 for (i, &sym) in state.labels.iter().enumerate() {
615 if sym == u32::MAX {
616 continue; }
618 let id = i as u32;
619 let value = cv.get(id, core_storage::NS_PROP).map(|vr| vr.into_value());
620 if core_storage::namespace_of_value(value.as_ref()) == namespace {
621 visible.insert(id);
622 }
623 }
624 Ok(NodeMask::from_ids(visible))
625 }
626
627 pub fn resolve_key(&self, key: &str) -> Option<u32> {
632 self.effective().ok()?.ids.get(key)
633 }
634
635 pub fn query(&self, cypher: &str, params: &BTreeMap<String, Value>) -> Result<ResultSet> {
637 let tokens = lex(cypher).map_err(|e| GraphError::QueryError {
638 detail: format!("lex: {e}"),
639 })?;
640 let ast = parse(&tokens).map_err(|e| GraphError::QueryError {
641 detail: format!("parse: {e}"),
642 })?;
643 let ops = plan(&ast).map_err(|e| GraphError::QueryError {
644 detail: format!("plan: {e}"),
645 })?;
646 let state = self.effective()?;
647 let view = make_view(state, &self.base, None);
648 execute(&view, &ops, &Params(params)).map_err(|e| GraphError::QueryError {
649 detail: format!("execute: {e}"),
650 })
651 }
652
653 pub fn query_masked(
657 &self,
658 cypher: &str,
659 params: &BTreeMap<String, Value>,
660 mask: &NodeMask,
661 ) -> Result<ResultSet> {
662 let tokens = lex(cypher).map_err(|e| GraphError::QueryError {
663 detail: format!("lex: {e}"),
664 })?;
665 if is_write_tokens(&tokens) {
666 return Err(GraphError::QueryError {
667 detail: "masked queries are read-only".into(),
668 });
669 }
670 let ast = parse(&tokens).map_err(|e| GraphError::QueryError {
671 detail: format!("parse: {e}"),
672 })?;
673 let ops = plan(&ast).map_err(|e| GraphError::QueryError {
674 detail: format!("plan: {e}"),
675 })?;
676 let state = self.effective()?;
677 let view = make_view(state, &self.base, Some(&mask.visible));
678 execute(&view, &ops, &Params(params)).map_err(|e| GraphError::QueryError {
679 detail: format!("execute: {e}"),
680 })
681 }
682
683 pub fn node_info(&self, key: &str) -> Option<NodeInfo> {
685 node_info_from(key, self.effective().ok()?, &self.base)
686 }
687
688 pub fn node_edges(&self, key: &str) -> Result<Vec<EdgeInfo>> {
693 node_edges_from(key, self.effective()?, &self.base)
694 }
695
696 pub fn neighborhood_masked(
701 &self,
702 key: &str,
703 depth: u32,
704 edge_types: Option<&[&str]>,
705 dir: Dir,
706 mask: &NodeMask,
707 ) -> Option<ResultSet> {
708 neighborhood_masked_from(
709 key,
710 self.effective().ok()?,
711 &self.base,
712 depth,
713 edge_types,
714 dir,
715 mask,
716 )
717 }
718
719 pub fn node_edges_scoped(&self, key: &str, mask: &NodeMask) -> Result<Vec<EdgeInfo>> {
733 let state = self.effective()?;
734 if !state.ids.get(key).is_some_and(|id| mask.contains_id(id)) {
735 return Err(GraphError::KeyNotFound { key: key.into() });
736 }
737 let edges = node_edges_from(key, state, &self.base)?;
738 Ok(edges
739 .into_iter()
740 .filter(|e| {
741 let other = if e.src_key == key {
742 &e.dst_key
743 } else {
744 &e.src_key
745 };
746 state.ids.get(other).is_some_and(|id| mask.contains_id(id))
747 })
748 .collect())
749 }
750
751 pub fn neighborhood_scoped(
761 &self,
762 key: &str,
763 depth: u32,
764 edge_types: Option<&[&str]>,
765 dir: Dir,
766 mask: &NodeMask,
767 ) -> Result<ResultSet> {
768 let state = self.effective()?;
769 if !state.ids.get(key).is_some_and(|id| mask.contains_id(id)) {
770 return Err(GraphError::KeyNotFound { key: key.into() });
771 }
772 neighborhood_masked_from(key, state, &self.base, depth, edge_types, dir, mask)
773 .ok_or_else(|| GraphError::KeyNotFound { key: key.into() })
774 }
775}
776
777fn node_info_from(
780 key: &str,
781 state: &FrozenOverlay,
782 base: &Option<Arc<MappedBase>>,
783) -> Option<NodeInfo> {
784 let id = state.ids.get(key)?;
785 let label_sym = *state.labels.get(id as usize)?;
786 if label_sym == u32::MAX {
787 return None;
788 }
789 let label = state.syms.resolve(label_sym)?.to_string();
790 let cv = build_cv(&state.props, base);
791 let mut props = BTreeMap::new();
792 for field in cv.field_names() {
793 if let Some(vr) = cv.get(id, &field) {
794 props.insert(field, vr.into_value());
795 }
796 }
797 Some(NodeInfo {
798 key: key.to_string(),
799 label,
800 props,
801 })
802}
803
804fn node_edges_from(
805 key: &str,
806 state: &FrozenOverlay,
807 base: &Option<Arc<MappedBase>>,
808) -> Result<Vec<EdgeInfo>> {
809 let id = state
810 .ids
811 .get(key)
812 .ok_or_else(|| GraphError::KeyNotFound { key: key.into() })?;
813 let tv = build_tv(&state.topo, base);
814 let mut edges = Vec::new();
815 for etype in tv.etypes() {
816 let edge_type = state
817 .syms
818 .resolve(etype)
819 .ok_or_else(|| GraphError::Corrupt {
820 detail: format!("reader: topology etype {etype} not in interner"),
821 })?
822 .to_string();
823 for dir in [Direction::Out, Direction::In] {
824 for &nbr in tv.neighbors(etype, dir, id).as_ref() {
825 let (src_key, dst_key) = match dir {
826 Direction::Out => (
827 key.to_string(),
828 state
829 .ids
830 .key_of(nbr)
831 .ok_or_else(|| GraphError::Corrupt {
832 detail: format!("topology id {nbr} has no key"),
833 })?
834 .to_string(),
835 ),
836 Direction::In => (
837 state
838 .ids
839 .key_of(nbr)
840 .ok_or_else(|| GraphError::Corrupt {
841 detail: format!("topology id {nbr} has no key"),
842 })?
843 .to_string(),
844 key.to_string(),
845 ),
846 };
847 edges.push(EdgeInfo {
848 edge_type: edge_type.clone(),
849 src_key,
850 dst_key,
851 derived: false,
852 });
853 }
854 }
855 }
856 edges.sort_by(|a, b| {
857 a.edge_type
858 .cmp(&b.edge_type)
859 .then(a.src_key.cmp(&b.src_key))
860 .then(a.dst_key.cmp(&b.dst_key))
861 });
862 edges.dedup();
863 Ok(edges)
864}
865
866fn neighborhood_masked_from(
867 key: &str,
868 state: &FrozenOverlay,
869 base: &Option<Arc<MappedBase>>,
870 depth: u32,
871 edge_types: Option<&[&str]>,
872 dir: Dir,
873 mask: &NodeMask,
874) -> Option<ResultSet> {
875 let start_id = state.ids.get(key)?;
876 let view = make_view(state, base, Some(&mask.visible));
877 let resolved: Option<Vec<u32>> = edge_types.map(|names| {
878 names
879 .iter()
880 .filter_map(|name| view.syms.get(name))
881 .collect()
882 });
883 let nb = neighborhood(&view, start_id, depth, resolved.as_deref(), dir);
884 let mut rs = ResultSet::new(vec!["key".into(), "label".into(), "depth".into()]);
885 let mut visited: Vec<(u32, u32)> = Vec::with_capacity(nb.nodes.len() + 1);
887 visited.push((start_id, 0));
888 for (nid, d) in &nb.nodes {
889 let k = view.key_of(*nid);
890 let lbl = view
891 .label_of(*nid)
892 .expect("real nodes always have a label; u32::MAX sentinel cannot occur");
893 rs.push_row(vec![
894 Some(Value::Str(k.to_string())),
895 Some(Value::Str(lbl.to_string())),
896 Some(Value::Int(*d as i64)),
897 ]);
898 visited.push((*nid, *d));
899 }
900 if mask.mode() == crate::mask::MaskMode::Stub {
908 let raw_view = make_view(state, base, None);
909 let mut seen: HashSet<u32> = visited.iter().map(|(id, _)| *id).collect();
910 for (node_id, node_depth) in &visited {
911 if *node_depth >= depth {
912 continue;
913 }
914 for e in expand(&raw_view, *node_id, resolved.as_deref(), dir) {
915 let nbr = if e.src == *node_id { e.dst } else { e.src };
916 if !mask.contains_id(nbr) && seen.insert(nbr) {
917 if let Some(k) = state.ids.key_of(nbr) {
918 rs.push_row(vec![
919 Some(Value::Str(k.to_string())),
920 None,
921 Some(Value::Int((*node_depth + 1) as i64)),
922 ]);
923 }
924 }
925 }
926 }
927 }
928 Some(rs)
929}