1use std::collections::{BTreeMap, BTreeSet};
2use std::path::Path;
3
4use marsdb_storage::{
5 ReadTransaction, ReadableMultimapTable, ReadableTable, StorageEngine, Txn, WriteTransaction,
6};
7
8use crate::encode::{
9 decode_edge, decode_node, edge_header, encode_edge, encode_node, node_label_ids, EdgeRecord,
10 NodeRecord,
11};
12use crate::error::GraphError;
13use crate::id::next_id;
14use crate::labels::{intern_label, lookup_label_id, resolve_label};
15use crate::model::{AdjEntry, Direction, Edge, EdgeId, Node, NodeId, PropertyValue};
16use crate::props::{intern_prop, prop_resolver};
17use crate::write_ctx::WriteCtx;
18
19pub struct GraphStore {
20 storage: StorageEngine,
21}
22
23#[derive(Debug, Clone, PartialEq, Eq)]
25pub struct IntegrityReport {
26 pub physical_was_clean: bool,
29 pub labels: u64,
30 pub nodes: u64,
31 pub edges: u64,
32}
33
34impl GraphStore {
35 pub fn open_file(path: impl AsRef<Path>) -> Result<Self, GraphError> {
36 Ok(Self {
37 storage: StorageEngine::open_file(path)?,
38 })
39 }
40
41 pub fn open_memory() -> Result<Self, GraphError> {
42 Ok(Self {
43 storage: StorageEngine::open_memory()?,
44 })
45 }
46
47 pub fn backup_to(&self, path: impl AsRef<Path>) -> Result<(), GraphError> {
48 self.storage.backup_to(path)?;
49 Ok(())
50 }
51
52 pub fn check_integrity(&mut self) -> Result<IntegrityReport, GraphError> {
55 let physical_was_clean = self.storage.check_integrity()?;
56 let read = self.storage.begin_read()?;
57
58 let mut labels_by_id = BTreeMap::new();
59 {
60 let table = read.open_table(marsdb_storage::tables::ID_TO_LABEL)?;
61 for entry in table.iter()? {
62 let (id, label) = entry?;
63 labels_by_id.insert(id.value(), label.value().to_owned());
64 }
65 }
66 {
67 let table = read.open_table(marsdb_storage::tables::LABEL_TO_ID)?;
68 let mut count = 0usize;
69 for entry in table.iter()? {
70 let (label, id) = entry?;
71 count += 1;
72 if labels_by_id.get(&id.value()).map(String::as_str) != Some(label.value()) {
73 return Err(GraphError::CorruptData(format!(
74 "label mapping {:?} -> {} has no matching reverse mapping",
75 label.value(),
76 id.value()
77 )));
78 }
79 }
80 if count != labels_by_id.len() {
81 return Err(GraphError::CorruptData(
82 "label mapping tables have different entry counts".into(),
83 ));
84 }
85 }
86
87 let mut nodes = BTreeMap::<u64, Vec<u32>>::new();
88 {
89 let table = read.open_table(marsdb_storage::tables::NODES)?;
90 for entry in table.iter()? {
91 let (id, value) = entry?;
92 let label_ids = node_label_ids(value.value())?;
95 for label_id in &label_ids {
96 if !labels_by_id.contains_key(label_id) {
97 return Err(GraphError::CorruptData(format!(
98 "node {} references unknown label {}",
99 id.value(),
100 label_id
101 )));
102 }
103 }
104 nodes.insert(id.value(), label_ids);
105 }
106 }
107
108 let mut indexed_labels = BTreeSet::new();
109 {
110 let table = read.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
111 for entry in table.iter()? {
112 let (label_id, values) = entry?;
113 let label_id = label_id.value();
114 if !labels_by_id.contains_key(&label_id) {
115 return Err(GraphError::CorruptData(format!(
116 "node label index references unknown label {label_id}"
117 )));
118 }
119 for node_id in values {
120 let node_id = node_id?.value();
121 let Some(node_labels) = nodes.get(&node_id) else {
122 return Err(GraphError::CorruptData(format!(
123 "node label index references missing node {node_id}"
124 )));
125 };
126 if !node_labels.contains(&label_id) {
127 return Err(GraphError::CorruptData(format!(
128 "node label index has label {label_id} for node {node_id}, but the node does not"
129 )));
130 }
131 indexed_labels.insert((label_id, node_id));
132 }
133 }
134 }
135 for (node_id, label_ids) in &nodes {
136 for label_id in label_ids {
137 if !indexed_labels.contains(&(*label_id, *node_id)) {
138 return Err(GraphError::CorruptData(format!(
139 "node {node_id} has label {label_id} but is missing from the label index"
140 )));
141 }
142 }
143 }
144
145 let mut edges = BTreeMap::<u64, (u32, u64, u64)>::new();
146 {
147 let table = read.open_table(marsdb_storage::tables::EDGES)?;
148 for entry in table.iter()? {
149 let (id, value) = entry?;
150 let (label_id, src, dst) = edge_header(value.value())?;
152 if !labels_by_id.contains_key(&label_id) {
153 return Err(GraphError::CorruptData(format!(
154 "edge {} references unknown label {}",
155 id.value(),
156 label_id
157 )));
158 }
159 if !nodes.contains_key(&src) || !nodes.contains_key(&dst) {
160 return Err(GraphError::CorruptData(format!(
161 "edge {} references missing endpoint {} -> {}",
162 id.value(),
163 src,
164 dst
165 )));
166 }
167 edges.insert(id.value(), (label_id, src, dst));
168 }
169 }
170
171 let outgoing =
172 Self::check_adjacency(&read, marsdb_storage::tables::ADJ_OUT, &nodes, &edges, true)?;
173 let incoming =
174 Self::check_adjacency(&read, marsdb_storage::tables::ADJ_IN, &nodes, &edges, false)?;
175 for (&edge_id, &(label_id, src, dst)) in &edges {
176 if !outgoing.contains(&(src, edge_id, dst, label_id)) {
177 return Err(GraphError::CorruptData(format!(
178 "edge {edge_id} is missing from outgoing adjacency"
179 )));
180 }
181 if !incoming.contains(&(dst, edge_id, src, label_id)) {
182 return Err(GraphError::CorruptData(format!(
183 "edge {edge_id} is missing from incoming adjacency"
184 )));
185 }
186 }
187
188 let meta = read.open_table(marsdb_storage::tables::META)?;
189 for (counter, maximum) in [
190 ("next_node_id", nodes.keys().next_back().copied()),
191 ("next_edge_id", edges.keys().next_back().copied()),
192 ] {
193 if let Some(maximum) = maximum {
194 let stored = meta.get(counter)?.map(|value| value.value()).unwrap_or(0);
195 if stored < maximum {
196 return Err(GraphError::CorruptData(format!(
197 "{counter} counter {stored} is below maximum allocated id {maximum}"
198 )));
199 }
200 }
201 }
202
203 Ok(IntegrityReport {
204 physical_was_clean,
205 labels: labels_by_id.len() as u64,
206 nodes: nodes.len() as u64,
207 edges: edges.len() as u64,
208 })
209 }
210
211 fn check_adjacency(
212 read: &ReadTransaction,
213 definition: marsdb_storage::TableDefinition<(u64, u32, u64), u64>,
214 nodes: &BTreeMap<u64, Vec<u32>>,
215 edges: &BTreeMap<u64, (u32, u64, u64)>,
216 outgoing: bool,
217 ) -> Result<BTreeSet<(u64, u64, u64, u32)>, GraphError> {
218 let table = read.open_table(definition)?;
219 let mut found = BTreeSet::new();
220 for entry in table.iter()? {
221 let (key, value) = entry?;
222 let (owner, key_label_id, edge_id) = key.value();
223 let other = value.value();
224 if !nodes.contains_key(&owner) {
225 return Err(GraphError::CorruptData(format!(
226 "adjacency references missing owner node {owner}"
227 )));
228 }
229 let Some(&(label_id, src, dst)) = edges.get(&edge_id) else {
230 return Err(GraphError::CorruptData(format!(
231 "adjacency references missing edge {edge_id}"
232 )));
233 };
234 let expected = if outgoing { (src, dst) } else { (dst, src) };
235 if owner != expected.0 || other != expected.1 || key_label_id != label_id {
236 return Err(GraphError::CorruptData(format!(
237 "adjacency entry for edge {edge_id} does not match the edge record"
238 )));
239 }
240 found.insert((owner, edge_id, other, key_label_id));
241 }
242 Ok(found)
243 }
244
245 pub fn begin_write(&self) -> Result<WriteTransaction, GraphError> {
260 Ok(self.storage.begin_write()?)
261 }
262
263 pub fn begin_read(&self) -> Result<ReadTransaction, GraphError> {
270 Ok(self.storage.begin_read()?)
271 }
272
273 pub fn commit(write_txn: WriteTransaction) -> Result<(), GraphError> {
275 write_txn.commit()?;
276 Ok(())
277 }
278
279 pub fn abort(write_txn: WriteTransaction) -> Result<(), GraphError> {
283 write_txn.abort()?;
284 Ok(())
285 }
286
287 pub fn create_node(
288 &self,
289 labels: &[&str],
290 props: BTreeMap<String, PropertyValue>,
291 ) -> Result<NodeId, GraphError> {
292 let write_txn = self.begin_write()?;
293 let id = Self::create_node_in_txn(&write_txn, labels, props)?;
294 write_txn.commit()?;
295 Ok(id)
296 }
297
298 pub fn create_node_in_txn(
299 write_txn: &WriteTransaction,
300 labels: &[&str],
301 props: BTreeMap<String, PropertyValue>,
302 ) -> Result<NodeId, GraphError> {
303 let mut ctx = WriteCtx::open(write_txn);
304 Self::create_node_ctx(&mut ctx, labels, props)
305 }
306
307 fn create_node_ctx(
308 ctx: &mut WriteCtx,
309 labels: &[&str],
310 props: BTreeMap<String, PropertyValue>,
311 ) -> Result<NodeId, GraphError> {
312 let label_ids = labels
313 .iter()
314 .map(|l| intern_label(ctx, l))
315 .collect::<Result<Vec<_>, _>>()?;
316 let id = next_id(ctx, "next_node_id")?;
317 let record = NodeRecord {
318 label_ids: label_ids.clone(),
319 props,
320 };
321 let bytes = encode_node(&record, |name| intern_prop(ctx, name))?;
322 ctx.nodes()?.insert(id, bytes.as_slice())?;
323 for &label_id in &label_ids {
324 ctx.node_label_index()?.insert(label_id, id)?;
325 }
326 crate::index::on_node_created(ctx, id, &label_ids, &record.props)?;
327 Ok(NodeId(id))
328 }
329
330 pub fn get_node(&self, id: NodeId) -> Result<Option<Node>, GraphError> {
331 let read_txn = self.begin_read()?;
332 Self::get_node_in_txn(Txn::Read(&read_txn), id)
333 }
334
335 pub fn get_node_in_txn(txn: Txn, id: NodeId) -> Result<Option<Node>, GraphError> {
336 let record: Option<NodeRecord> = {
337 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
338 let found = match nodes.get(id.0)? {
339 Some(guard) => {
340 let mut resolve = prop_resolver(txn)?;
341 Some(decode_node(guard.value(), &mut resolve)?)
342 }
343 None => None,
344 };
345 found
346 };
347 let Some(record) = record else {
348 return Ok(None);
349 };
350 let labels = record
351 .label_ids
352 .iter()
353 .map(|&lid| resolve_label(txn, lid))
354 .collect::<Result<Vec<_>, _>>()?;
355 Ok(Some(Node {
356 id,
357 labels,
358 props: record.props,
359 }))
360 }
361
362 pub fn lookup_prop_id_in_txn(txn: Txn, prop: &str) -> Result<Option<u32>, GraphError> {
368 crate::props::lookup_prop_id(txn, prop)
369 }
370
371 pub fn get_node_prop_in_txn(
381 txn: Txn,
382 id: NodeId,
383 prop_id: u32,
384 ) -> Result<Option<Option<PropertyValue>>, GraphError> {
385 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
386 let Some(guard) = nodes.get(id.0)? else {
387 return Ok(None);
388 };
389 match crate::encode::node_prop_raw(guard.value(), prop_id)? {
390 Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
391 None => Ok(Some(None)),
392 }
393 }
394
395 pub fn get_edge_prop_in_txn(
398 txn: Txn,
399 id: EdgeId,
400 prop_id: u32,
401 ) -> Result<Option<Option<PropertyValue>>, GraphError> {
402 let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
403 let Some(guard) = edges.get(id.0)? else {
404 return Ok(None);
405 };
406 match crate::encode::edge_prop_raw(guard.value(), prop_id)? {
407 Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
408 None => Ok(Some(None)),
409 }
410 }
411
412 #[allow(clippy::type_complexity)] pub fn node_prop_reader(
419 txn: Txn<'_>,
420 ) -> Result<
421 impl FnMut(NodeId, u32) -> Result<Option<Option<PropertyValue>>, GraphError> + '_,
422 GraphError,
423 > {
424 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
425 Ok(move |id: NodeId, prop_id: u32| {
426 let Some(guard) = nodes.get(id.0)? else {
427 return Ok(None);
428 };
429 match crate::encode::node_prop_raw(guard.value(), prop_id)? {
430 Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
431 None => Ok(Some(None)),
432 }
433 })
434 }
435
436 pub fn node_exists_in_txn(txn: Txn, id: NodeId) -> Result<bool, GraphError> {
441 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
442 let exists = nodes.get(id.0)?.is_some();
443 Ok(exists)
444 }
445
446 pub fn edge_exists_in_txn(txn: Txn, id: EdgeId) -> Result<bool, GraphError> {
448 let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
449 let exists = edges.get(id.0)?.is_some();
450 Ok(exists)
451 }
452
453 pub fn create_edge(
454 &self,
455 label: &str,
456 src: NodeId,
457 dst: NodeId,
458 props: BTreeMap<String, PropertyValue>,
459 ) -> Result<EdgeId, GraphError> {
460 let write_txn = self.begin_write()?;
461 let id = Self::create_edge_in_txn(&write_txn, label, src, dst, props)?;
462 write_txn.commit()?;
463 Ok(id)
464 }
465
466 pub fn create_edge_in_txn(
467 write_txn: &WriteTransaction,
468 label: &str,
469 src: NodeId,
470 dst: NodeId,
471 props: BTreeMap<String, PropertyValue>,
472 ) -> Result<EdgeId, GraphError> {
473 let mut ctx = WriteCtx::open(write_txn);
474 Self::create_edge_ctx(&mut ctx, label, src, dst, props)
475 }
476
477 fn create_edge_ctx(
478 ctx: &mut WriteCtx,
479 label: &str,
480 src: NodeId,
481 dst: NodeId,
482 props: BTreeMap<String, PropertyValue>,
483 ) -> Result<EdgeId, GraphError> {
484 if ctx.nodes()?.get(src.0)?.is_none() {
485 return Err(GraphError::NodeNotFound(src));
486 }
487 if ctx.nodes()?.get(dst.0)?.is_none() {
488 return Err(GraphError::NodeNotFound(dst));
489 }
490 let label_id = intern_label(ctx, label)?;
491 let id = next_id(ctx, "next_edge_id")?;
492 let record = EdgeRecord {
493 label_id,
494 src: src.0,
495 dst: dst.0,
496 props,
497 };
498 let bytes = encode_edge(&record, |name| intern_prop(ctx, name))?;
499 ctx.edges()?.insert(id, bytes.as_slice())?;
500
501 ctx.adj_out()?
502 .insert(crate::model::adj_key(src.0, label_id, id), dst.0)?;
503 ctx.adj_in()?
504 .insert(crate::model::adj_key(dst.0, label_id, id), src.0)?;
505 Ok(EdgeId(id))
506 }
507
508 pub fn get_edge(&self, id: EdgeId) -> Result<Option<Edge>, GraphError> {
509 let read_txn = self.begin_read()?;
510 Self::get_edge_in_txn(Txn::Read(&read_txn), id)
511 }
512
513 pub fn get_edge_in_txn(txn: Txn, id: EdgeId) -> Result<Option<Edge>, GraphError> {
514 let record: Option<EdgeRecord> = {
515 let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
516 let found = match edges.get(id.0)? {
517 Some(guard) => {
518 let mut resolve = prop_resolver(txn)?;
519 Some(decode_edge(guard.value(), &mut resolve)?)
520 }
521 None => None,
522 };
523 found
524 };
525 let Some(record) = record else {
526 return Ok(None);
527 };
528 let label = resolve_label(txn, record.label_id)?;
529 Ok(Some(Edge {
530 id,
531 label,
532 src: NodeId(record.src),
533 dst: NodeId(record.dst),
534 props: record.props,
535 }))
536 }
537
538 pub fn neighbors(
541 &self,
542 node: NodeId,
543 dir: Direction,
544 label_filter: Option<&str>,
545 ) -> Result<Vec<AdjEntry>, GraphError> {
546 let read_txn = self.begin_read()?;
547 Self::neighbors_in_txn(Txn::Read(&read_txn), node, dir, label_filter)
548 }
549
550 pub fn neighbors_in_txn(
551 txn: Txn,
552 node: NodeId,
553 dir: Direction,
554 label_filter: Option<&str>,
555 ) -> Result<Vec<AdjEntry>, GraphError> {
556 let (lo, hi) = match label_filter {
560 Some(l) => match lookup_label_id(txn, l)? {
561 Some(lid) => crate::model::adj_label_bounds(node.0, lid),
562 None => return Ok(Vec::new()),
563 },
564 None => crate::model::adj_node_bounds(node.0),
565 };
566 let mut result = Vec::new();
567 let table_def = match dir {
568 Direction::Out => marsdb_storage::tables::ADJ_OUT,
569 Direction::In => marsdb_storage::tables::ADJ_IN,
570 };
571 let table = txn.open_table(table_def)?;
572 for item in table.range(lo..=hi)? {
573 let (key, value) = item?;
574 let (_, label_id, edge_id) = key.value();
575 result.push(AdjEntry {
576 edge_id: EdgeId(edge_id),
577 other: NodeId(value.value()),
578 label_id,
579 });
580 }
581 Ok(result)
582 }
583
584 pub fn delete_edge(&self, id: EdgeId) -> Result<bool, GraphError> {
585 let write_txn = self.begin_write()?;
586 let removed = Self::delete_edge_in_txn(&write_txn, id)?;
587 write_txn.commit()?;
588 Ok(removed)
589 }
590
591 pub fn delete_edge_in_txn(
592 write_txn: &WriteTransaction,
593 id: EdgeId,
594 ) -> Result<bool, GraphError> {
595 let mut ctx = WriteCtx::open(write_txn);
596 Self::delete_edge_ctx(&mut ctx, id)
597 }
598
599 fn delete_edge_ctx(ctx: &mut WriteCtx, id: EdgeId) -> Result<bool, GraphError> {
606 let Some(record_bytes) = ctx
607 .edges()?
608 .remove(id.0)?
609 .map(|guard| guard.value().to_vec())
610 else {
611 return Ok(false);
612 };
613 let (label_id, src, dst) = edge_header(&record_bytes)?;
616 ctx.adj_out()?
617 .remove(crate::model::adj_key(src, label_id, id.0))?;
618 ctx.adj_in()?
619 .remove(crate::model::adj_key(dst, label_id, id.0))?;
620 Ok(true)
621 }
622
623 pub fn delete_node(&self, id: NodeId, detach: bool) -> Result<bool, GraphError> {
626 let write_txn = self.begin_write()?;
627 let existed = Self::delete_node_in_txn(&write_txn, id, detach)?;
628 write_txn.commit()?;
629 Ok(existed)
630 }
631
632 pub fn delete_node_in_txn(
633 write_txn: &WriteTransaction,
634 id: NodeId,
635 detach: bool,
636 ) -> Result<bool, GraphError> {
637 let mut ctx = WriteCtx::open(write_txn);
638 let mut incident: Vec<EdgeId> = Vec::new();
639 let (lo, hi) = crate::model::adj_node_bounds(id.0);
640 for item in ctx.adj_out()?.range(lo..=hi)? {
641 let (key, _) = item?;
642 let (_, _, edge_id) = key.value();
643 incident.push(EdgeId(edge_id));
644 }
645 for item in ctx.adj_in()?.range(lo..=hi)? {
646 let (key, _) = item?;
647 let (_, _, edge_id) = key.value();
648 incident.push(EdgeId(edge_id));
649 }
650 if !incident.is_empty() && !detach {
651 return Err(GraphError::NodeHasEdges(id));
652 }
653 for edge_id in incident {
654 Self::delete_edge_ctx(&mut ctx, edge_id)?;
655 }
656 let Some(removed_bytes) = ctx
657 .nodes()?
658 .remove(id.0)?
659 .map(|guard| guard.value().to_vec())
660 else {
661 return Ok(false);
662 };
663 let record = decode_node(&removed_bytes, |pid| {
664 crate::index::resolve_prop_ctx(&mut ctx, pid)
665 })?;
666 for &label_id in &record.label_ids {
667 ctx.node_label_index()?.remove(label_id, id.0)?;
668 }
669 crate::index::on_node_deleted(&mut ctx, id.0, &record.label_ids, &record.props)?;
670 Ok(true)
671 }
672
673 pub fn create_index(&self, label: &str, prop: &str, unique: bool) -> Result<(), GraphError> {
677 let write_txn = self.begin_write()?;
678 Self::create_index_in_txn(&write_txn, label, prop, unique)?;
679 write_txn.commit()?;
680 Ok(())
681 }
682
683 pub fn create_index_in_txn(
689 write_txn: &WriteTransaction,
690 label: &str,
691 prop: &str,
692 unique: bool,
693 ) -> Result<(), GraphError> {
694 let mut ctx = WriteCtx::open(write_txn);
695 crate::index::create_index(&mut ctx, label, prop, unique)
696 }
697
698 pub fn index_def(
700 &self,
701 label: &str,
702 prop: &str,
703 ) -> Result<Option<crate::IndexDef>, GraphError> {
704 let read_txn = self.begin_read()?;
705 crate::index::lookup_index_def(Txn::Read(&read_txn), label, prop)
706 }
707
708 pub fn index_def_in_txn(
712 txn: Txn,
713 label: &str,
714 prop: &str,
715 ) -> Result<Option<crate::IndexDef>, GraphError> {
716 crate::index::lookup_index_def(txn, label, prop)
717 }
718
719 pub fn lookup_by_index_in_txn(
721 txn: Txn,
722 label: &str,
723 prop: &str,
724 value: &PropertyValue,
725 ) -> Result<Vec<NodeId>, GraphError> {
726 crate::index::lookup_exact(txn, label, prop, value, None)
727 }
728
729 pub fn lookup_by_index_limited_in_txn(
733 txn: Txn,
734 label: &str,
735 prop: &str,
736 value: &PropertyValue,
737 limit: usize,
738 ) -> Result<Vec<NodeId>, GraphError> {
739 crate::index::lookup_exact(txn, label, prop, value, Some(limit))
740 }
741
742 pub fn index_match_count_in_txn(
748 txn: Txn,
749 label: &str,
750 prop: &str,
751 value: &PropertyValue,
752 ) -> Result<u64, GraphError> {
753 crate::index::match_count(txn, label, prop, value)
754 }
755
756 pub fn lookup_by_index(
760 &self,
761 label: &str,
762 prop: &str,
763 value: &PropertyValue,
764 ) -> Result<Vec<NodeId>, GraphError> {
765 let read_txn = self.begin_read()?;
766 crate::index::lookup_exact(Txn::Read(&read_txn), label, prop, value, None)
767 }
768
769 pub fn set_node_prop(
770 &self,
771 id: NodeId,
772 key: &str,
773 value: PropertyValue,
774 ) -> Result<bool, GraphError> {
775 let write_txn = self.begin_write()?;
776 let updated = Self::set_node_prop_in_txn(&write_txn, id, key, value)?;
777 if updated {
778 write_txn.commit()?;
779 } else {
780 write_txn.abort()?;
781 }
782 Ok(updated)
783 }
784
785 pub fn set_node_prop_in_txn(
786 write_txn: &WriteTransaction,
787 id: NodeId,
788 key: &str,
789 value: PropertyValue,
790 ) -> Result<bool, GraphError> {
791 let mut ctx = WriteCtx::open(write_txn);
792 let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
793 return Ok(false);
794 };
795 let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
796 let old_value = record.props.insert(key.to_string(), value.clone());
797 let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
798 ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
799 crate::index::on_node_prop_changed(
800 &mut ctx,
801 id.0,
802 &record.label_ids,
803 key,
804 old_value.as_ref(),
805 Some(&value),
806 )?;
807 Ok(true)
808 }
809
810 pub fn set_edge_prop(
811 &self,
812 id: EdgeId,
813 key: &str,
814 value: PropertyValue,
815 ) -> Result<bool, GraphError> {
816 let write_txn = self.begin_write()?;
817 let updated = Self::set_edge_prop_in_txn(&write_txn, id, key, value)?;
818 if updated {
819 write_txn.commit()?;
820 } else {
821 write_txn.abort()?;
822 }
823 Ok(updated)
824 }
825
826 pub fn set_edge_prop_in_txn(
827 write_txn: &WriteTransaction,
828 id: EdgeId,
829 key: &str,
830 value: PropertyValue,
831 ) -> Result<bool, GraphError> {
832 let mut ctx = WriteCtx::open(write_txn);
833 let Some(bytes) = ctx.edges()?.get(id.0)?.map(|g| g.value().to_vec()) else {
834 return Ok(false);
835 };
836 let mut record = decode_edge(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
837 record.props.insert(key.to_string(), value);
838 let new_bytes = encode_edge(&record, |name| intern_prop(&mut ctx, name))?;
839 ctx.edges()?.insert(id.0, new_bytes.as_slice())?;
840 Ok(true)
841 }
842
843 pub fn remove_node_prop_in_txn(
844 write_txn: &WriteTransaction,
845 id: NodeId,
846 key: &str,
847 ) -> Result<bool, GraphError> {
848 let mut ctx = WriteCtx::open(write_txn);
849 let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
850 return Ok(false);
851 };
852 let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
853 let old_value = record.props.remove(key);
854 let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
855 ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
856 crate::index::on_node_prop_changed(
857 &mut ctx,
858 id.0,
859 &record.label_ids,
860 key,
861 old_value.as_ref(),
862 None,
863 )?;
864 Ok(true)
865 }
866
867 pub fn remove_edge_prop_in_txn(
868 write_txn: &WriteTransaction,
869 id: EdgeId,
870 key: &str,
871 ) -> Result<bool, GraphError> {
872 let mut ctx = WriteCtx::open(write_txn);
873 let Some(bytes) = ctx.edges()?.get(id.0)?.map(|g| g.value().to_vec()) else {
874 return Ok(false);
875 };
876 let mut record = decode_edge(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
877 record.props.remove(key);
878 let new_bytes = encode_edge(&record, |name| intern_prop(&mut ctx, name))?;
879 ctx.edges()?.insert(id.0, new_bytes.as_slice())?;
880 Ok(true)
881 }
882
883 pub fn add_node_label_in_txn(
887 write_txn: &WriteTransaction,
888 id: NodeId,
889 label: &str,
890 ) -> Result<bool, GraphError> {
891 let mut ctx = WriteCtx::open(write_txn);
892 let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
893 return Ok(false);
894 };
895 let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
896 let label_id = intern_label(&mut ctx, label)?;
897 if !record.label_ids.contains(&label_id) {
898 record.label_ids.push(label_id);
899 let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
900 ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
901 ctx.node_label_index()?.insert(label_id, id.0)?;
902 crate::index::on_node_created(&mut ctx, id.0, &[label_id], &record.props)?;
903 }
904 Ok(true)
905 }
906
907 pub fn remove_node_label_in_txn(
911 write_txn: &WriteTransaction,
912 id: NodeId,
913 label: &str,
914 ) -> Result<bool, GraphError> {
915 let mut ctx = WriteCtx::open(write_txn);
916 let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
917 return Ok(false);
918 };
919 let Some(label_id) = ctx.label_to_id()?.get(label)?.map(|g| g.value()) else {
923 return Ok(true);
924 };
925 let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
926 if let Some(pos) = record.label_ids.iter().position(|&l| l == label_id) {
927 record.label_ids.remove(pos);
928 let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
929 ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
930 ctx.node_label_index()?.remove(label_id, id.0)?;
931 crate::index::on_node_deleted(&mut ctx, id.0, &[label_id], &record.props)?;
932 }
933 Ok(true)
934 }
935
936 pub fn node_count_in_txn(txn: Txn) -> Result<u64, GraphError> {
940 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
941 Ok(nodes.len()?)
942 }
943
944 pub fn label_count_in_txn(txn: Txn, label: &str) -> Result<u64, GraphError> {
949 let Some(label_id) = lookup_label_id(txn, label)? else {
950 return Ok(0);
951 };
952 let index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
953 let count = index.get(label_id)?.len();
954 Ok(count)
955 }
956
957 pub fn all_nodes(&self, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
960 let read_txn = self.begin_read()?;
961 Self::all_nodes_in_txn(Txn::Read(&read_txn), label_filter)
962 }
963
964 pub fn all_node_ids_limited_in_txn(
968 txn: Txn,
969 label_filter: Option<&str>,
970 limit: usize,
971 ) -> Result<Vec<NodeId>, GraphError> {
972 let Some(label_filter) = label_filter else {
973 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
974 return nodes
975 .iter()?
976 .take(limit)
977 .map(|entry| {
978 entry
979 .map(|(key, _)| NodeId(key.value()))
980 .map_err(Into::into)
981 })
982 .collect();
983 };
984 let Some(label_id) = lookup_label_id(txn, label_filter)? else {
985 return Ok(Vec::new());
986 };
987 let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
988 let ids = label_index
989 .get(label_id)?
990 .take(limit)
991 .map(|entry| entry.map(|value| NodeId(value.value())).map_err(Into::into))
992 .collect::<Result<Vec<_>, GraphError>>()?;
993 drop(label_index);
994 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
995 for id in &ids {
996 if nodes.get(id.0)?.is_none() {
997 return Err(GraphError::CorruptData(format!(
998 "node label index references missing node {}",
999 id.0
1000 )));
1001 }
1002 }
1003 Ok(ids)
1004 }
1005
1006 pub fn all_nodes_in_txn(txn: Txn, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
1007 Self::all_nodes_limited_in_txn(txn, label_filter, usize::MAX)
1008 }
1009
1010 pub fn all_nodes_limited_in_txn(
1017 txn: Txn,
1018 label_filter: Option<&str>,
1019 limit: usize,
1020 ) -> Result<Vec<Node>, GraphError> {
1021 let Some(label_filter) = label_filter else {
1027 let mut result = Vec::new();
1028 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
1029 let mut resolve = prop_resolver(txn)?;
1033 for item in nodes.iter()? {
1034 if result.len() >= limit {
1035 break;
1036 }
1037 let (key, value) = item?;
1038 let record = decode_node(value.value(), &mut resolve)?;
1039 let labels = record
1040 .label_ids
1041 .iter()
1042 .map(|&lid| resolve_label(txn, lid))
1043 .collect::<Result<Vec<_>, _>>()?;
1044 result.push(Node {
1045 id: NodeId(key.value()),
1046 labels,
1047 props: record.props,
1048 });
1049 }
1050 return Ok(result);
1051 };
1052 let Some(label_id) = lookup_label_id(txn, label_filter)? else {
1053 return Ok(Vec::new());
1054 };
1055 let node_ids: Vec<u64> = {
1056 let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
1057 let ids: Vec<u64> = label_index
1064 .get(label_id)?
1065 .take(limit)
1066 .map(|item| item.map(|g| g.value()))
1067 .collect::<Result<_, _>>()?;
1068 ids
1069 };
1070 let mut result = Vec::with_capacity(node_ids.len());
1071 let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
1072 let mut resolve = prop_resolver(txn)?;
1074 for id in node_ids {
1075 let guard = nodes.get(id)?.ok_or_else(|| {
1076 GraphError::CorruptData(format!("node label index references missing node {}", id))
1077 })?;
1078 let record = decode_node(guard.value(), &mut resolve)?;
1079 drop(guard);
1080 let labels = record
1081 .label_ids
1082 .iter()
1083 .map(|&lid| resolve_label(txn, lid))
1084 .collect::<Result<Vec<_>, _>>()?;
1085 result.push(Node {
1086 id: NodeId(id),
1087 labels,
1088 props: record.props,
1089 });
1090 }
1091 Ok(result)
1092 }
1093}
1094
1095#[cfg(test)]
1096mod tests {
1097 use super::*;
1098
1099 #[test]
1100 fn integrity_check_rejects_missing_node_label_index_entry() {
1101 let mut store = GraphStore::open_memory().unwrap();
1102 let node = store.create_node(&["Person"], BTreeMap::new()).unwrap();
1103
1104 let write = store.begin_write().unwrap();
1105 let label_id = {
1106 let labels = write
1107 .open_table(marsdb_storage::tables::LABEL_TO_ID)
1108 .unwrap();
1109 let id = labels.get("Person").unwrap().unwrap().value();
1110 id
1111 };
1112 write
1113 .open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)
1114 .unwrap()
1115 .remove(label_id, node.0)
1116 .unwrap();
1117 write.commit().unwrap();
1118
1119 let error = store.check_integrity().unwrap_err();
1120 assert!(
1121 matches!(error, GraphError::CorruptData(message) if message.contains("missing from the label index"))
1122 );
1123 }
1124
1125 #[test]
1126 fn integrity_check_rejects_dangling_adjacency_entry() {
1127 let mut store = GraphStore::open_memory().unwrap();
1128 let node = store.create_node(&[], BTreeMap::new()).unwrap();
1129
1130 let write = store.begin_write().unwrap();
1131 write
1132 .open_table(marsdb_storage::tables::ADJ_OUT)
1133 .unwrap()
1134 .insert(crate::model::adj_key(node.0, 0, 999), node.0)
1135 .unwrap();
1136 write.commit().unwrap();
1137
1138 let error = store.check_integrity().unwrap_err();
1139 assert!(
1140 matches!(error, GraphError::CorruptData(message) if message.contains("missing edge 999"))
1141 );
1142 }
1143
1144 #[test]
1145 fn create_index_backfills_existing_nodes() {
1146 let store = GraphStore::open_memory().unwrap();
1147 let mut alice_props = BTreeMap::new();
1148 alice_props.insert(
1149 "email".to_string(),
1150 PropertyValue::String("alice@x.com".to_string()),
1151 );
1152 let alice = store.create_node(&["Person"], alice_props).unwrap();
1153 let mut bob_props = BTreeMap::new();
1154 bob_props.insert(
1155 "email".to_string(),
1156 PropertyValue::String("bob@x.com".to_string()),
1157 );
1158 store.create_node(&["Person"], bob_props).unwrap();
1159 store.create_node(&["Person"], BTreeMap::new()).unwrap();
1161
1162 store.create_index("Person", "email", false).unwrap();
1163
1164 let found = store
1165 .lookup_by_index(
1166 "Person",
1167 "email",
1168 &PropertyValue::String("alice@x.com".to_string()),
1169 )
1170 .unwrap();
1171 assert_eq!(found, vec![alice]);
1172 }
1173
1174 #[test]
1175 fn create_index_rejects_duplicate_unique_value() {
1176 let store = GraphStore::open_memory().unwrap();
1177 let mut props1 = BTreeMap::new();
1178 props1.insert(
1179 "email".to_string(),
1180 PropertyValue::String("same@x.com".to_string()),
1181 );
1182 store.create_node(&["Person"], props1).unwrap();
1183 let mut props2 = BTreeMap::new();
1184 props2.insert(
1185 "email".to_string(),
1186 PropertyValue::String("same@x.com".to_string()),
1187 );
1188 store.create_node(&["Person"], props2).unwrap();
1189
1190 let error = store.create_index("Person", "email", true).unwrap_err();
1191 assert!(matches!(
1192 error,
1193 GraphError::UniqueConstraintViolation { .. }
1194 ));
1195
1196 assert!(store.index_def("Person", "email").unwrap().is_none());
1198 }
1199
1200 #[test]
1201 fn lookup_by_index_on_undeclared_index_is_empty_not_an_error() {
1202 let store = GraphStore::open_memory().unwrap();
1203 store.create_node(&["Person"], BTreeMap::new()).unwrap();
1204 let found = store
1205 .lookup_by_index("Person", "email", &PropertyValue::String("x".to_string()))
1206 .unwrap();
1207 assert_eq!(found, Vec::new());
1208 assert!(store.index_def("Person", "email").unwrap().is_none());
1209 }
1210
1211 #[test]
1212 fn index_survives_reopen() {
1213 let dir = tempfile::tempdir().unwrap();
1214 let path = dir.path().join("index.db");
1215 {
1216 let store = GraphStore::open_file(&path).unwrap();
1217 let mut props = BTreeMap::new();
1218 props.insert(
1219 "email".to_string(),
1220 PropertyValue::String("x@x.com".to_string()),
1221 );
1222 store.create_node(&["Person"], props).unwrap();
1223 store.create_index("Person", "email", false).unwrap();
1224 }
1225 let store = GraphStore::open_file(&path).unwrap();
1226 assert!(store.index_def("Person", "email").unwrap().is_some());
1227 let found = store
1228 .lookup_by_index(
1229 "Person",
1230 "email",
1231 &PropertyValue::String("x@x.com".to_string()),
1232 )
1233 .unwrap();
1234 assert_eq!(found.len(), 1);
1235 }
1236
1237 #[test]
1238 fn create_node_after_index_declared_is_indexed_immediately() {
1239 let store = GraphStore::open_memory().unwrap();
1240 store.create_index("Person", "email", false).unwrap();
1241 let mut props = BTreeMap::new();
1242 props.insert(
1243 "email".to_string(),
1244 PropertyValue::String("new@x.com".to_string()),
1245 );
1246 let node = store.create_node(&["Person"], props).unwrap();
1247
1248 let found = store
1249 .lookup_by_index(
1250 "Person",
1251 "email",
1252 &PropertyValue::String("new@x.com".to_string()),
1253 )
1254 .unwrap();
1255 assert_eq!(found, vec![node]);
1256 }
1257
1258 #[test]
1259 fn set_node_prop_moves_the_index_entry() {
1260 let store = GraphStore::open_memory().unwrap();
1261 let mut props = BTreeMap::new();
1262 props.insert(
1263 "email".to_string(),
1264 PropertyValue::String("old@x.com".to_string()),
1265 );
1266 let node = store.create_node(&["Person"], props).unwrap();
1267 store.create_index("Person", "email", false).unwrap();
1268
1269 store
1270 .set_node_prop(
1271 node,
1272 "email",
1273 PropertyValue::String("new@x.com".to_string()),
1274 )
1275 .unwrap();
1276
1277 assert!(store
1278 .lookup_by_index(
1279 "Person",
1280 "email",
1281 &PropertyValue::String("old@x.com".to_string())
1282 )
1283 .unwrap()
1284 .is_empty());
1285 assert_eq!(
1286 store
1287 .lookup_by_index(
1288 "Person",
1289 "email",
1290 &PropertyValue::String("new@x.com".to_string())
1291 )
1292 .unwrap(),
1293 vec![node]
1294 );
1295 }
1296
1297 #[test]
1298 fn set_node_prop_enforces_unique_index() {
1299 let store = GraphStore::open_memory().unwrap();
1300 let mut props1 = BTreeMap::new();
1301 props1.insert(
1302 "email".to_string(),
1303 PropertyValue::String("a@x.com".to_string()),
1304 );
1305 store.create_node(&["Person"], props1).unwrap();
1306 let mut props2 = BTreeMap::new();
1307 props2.insert(
1308 "email".to_string(),
1309 PropertyValue::String("b@x.com".to_string()),
1310 );
1311 let node2 = store.create_node(&["Person"], props2).unwrap();
1312 store.create_index("Person", "email", true).unwrap();
1313
1314 let error = store
1315 .set_node_prop(node2, "email", PropertyValue::String("a@x.com".to_string()))
1316 .unwrap_err();
1317 assert!(matches!(
1318 error,
1319 GraphError::UniqueConstraintViolation { .. }
1320 ));
1321 }
1322
1323 #[test]
1324 fn remove_node_prop_removes_the_index_entry() {
1325 let store = GraphStore::open_memory().unwrap();
1326 let mut props = BTreeMap::new();
1327 props.insert(
1328 "email".to_string(),
1329 PropertyValue::String("gone@x.com".to_string()),
1330 );
1331 let node = store.create_node(&["Person"], props).unwrap();
1332 store.create_index("Person", "email", false).unwrap();
1333
1334 let write = store.begin_write().unwrap();
1335 GraphStore::remove_node_prop_in_txn(&write, node, "email").unwrap();
1336 write.commit().unwrap();
1337
1338 assert!(store
1339 .lookup_by_index(
1340 "Person",
1341 "email",
1342 &PropertyValue::String("gone@x.com".to_string())
1343 )
1344 .unwrap()
1345 .is_empty());
1346 }
1347
1348 #[test]
1349 fn delete_node_removes_its_index_entries() {
1350 let store = GraphStore::open_memory().unwrap();
1351 let mut props = BTreeMap::new();
1352 props.insert(
1353 "email".to_string(),
1354 PropertyValue::String("deleted@x.com".to_string()),
1355 );
1356 let node = store.create_node(&["Person"], props).unwrap();
1357 store.create_index("Person", "email", false).unwrap();
1358
1359 store.delete_node(node, false).unwrap();
1360
1361 assert!(store
1362 .lookup_by_index(
1363 "Person",
1364 "email",
1365 &PropertyValue::String("deleted@x.com".to_string())
1366 )
1367 .unwrap()
1368 .is_empty());
1369 }
1370
1371 #[test]
1372 fn add_node_label_indexes_existing_props_under_the_new_label() {
1373 let store = GraphStore::open_memory().unwrap();
1374 let mut props = BTreeMap::new();
1375 props.insert(
1376 "email".to_string(),
1377 PropertyValue::String("multi@x.com".to_string()),
1378 );
1379 let node = store.create_node(&["Contact"], props).unwrap();
1380 store.create_index("Person", "email", false).unwrap();
1381
1382 assert!(store
1384 .lookup_by_index(
1385 "Person",
1386 "email",
1387 &PropertyValue::String("multi@x.com".to_string())
1388 )
1389 .unwrap()
1390 .is_empty());
1391
1392 let write = store.begin_write().unwrap();
1393 GraphStore::add_node_label_in_txn(&write, node, "Person").unwrap();
1394 write.commit().unwrap();
1395
1396 assert_eq!(
1397 store
1398 .lookup_by_index(
1399 "Person",
1400 "email",
1401 &PropertyValue::String("multi@x.com".to_string())
1402 )
1403 .unwrap(),
1404 vec![node]
1405 );
1406 }
1407
1408 #[test]
1409 fn remove_node_label_removes_index_entries_under_that_label() {
1410 let store = GraphStore::open_memory().unwrap();
1411 let mut props = BTreeMap::new();
1412 props.insert(
1413 "email".to_string(),
1414 PropertyValue::String("dual@x.com".to_string()),
1415 );
1416 let node = store.create_node(&["Person", "Contact"], props).unwrap();
1417 store.create_index("Person", "email", false).unwrap();
1418 assert_eq!(
1419 store
1420 .lookup_by_index(
1421 "Person",
1422 "email",
1423 &PropertyValue::String("dual@x.com".to_string())
1424 )
1425 .unwrap(),
1426 vec![node]
1427 );
1428
1429 let write = store.begin_write().unwrap();
1430 GraphStore::remove_node_label_in_txn(&write, node, "Person").unwrap();
1431 write.commit().unwrap();
1432
1433 assert!(store
1434 .lookup_by_index(
1435 "Person",
1436 "email",
1437 &PropertyValue::String("dual@x.com".to_string())
1438 )
1439 .unwrap()
1440 .is_empty());
1441 }
1442}