marsdb_graph/
store.rs

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::{decode, encode, EdgeRecord, NodeRecord};
9use crate::error::GraphError;
10use crate::id::next_id;
11use crate::labels::{intern_label, lookup_label_id, resolve_label};
12use crate::model::{AdjEntry, Direction, Edge, EdgeId, Node, NodeId, PropertyValue};
13
14pub struct GraphStore {
15    storage: StorageEngine,
16}
17
18/// Successful physical and logical integrity-check summary.
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct IntegrityReport {
21    /// `false` means redb detected physical damage and repaired it before
22    /// MarsDB's logical checks ran.
23    pub physical_was_clean: bool,
24    pub labels: u64,
25    pub nodes: u64,
26    pub edges: u64,
27}
28
29impl GraphStore {
30    pub fn open_file(path: impl AsRef<Path>) -> Result<Self, GraphError> {
31        Ok(Self {
32            storage: StorageEngine::open_file(path)?,
33        })
34    }
35
36    pub fn open_memory() -> Result<Self, GraphError> {
37        Ok(Self {
38            storage: StorageEngine::open_memory()?,
39        })
40    }
41
42    pub fn backup_to(&self, path: impl AsRef<Path>) -> Result<(), GraphError> {
43        self.storage.backup_to(path)?;
44        Ok(())
45    }
46
47    /// Check physical storage plus MarsDB's graph invariants. This requires
48    /// exclusive mutable access because redb may repair physical metadata.
49    pub fn check_integrity(&mut self) -> Result<IntegrityReport, GraphError> {
50        let physical_was_clean = self.storage.check_integrity()?;
51        let read = self.storage.begin_read()?;
52
53        let mut labels_by_id = BTreeMap::new();
54        {
55            let table = read.open_table(marsdb_storage::tables::ID_TO_LABEL)?;
56            for entry in table.iter()? {
57                let (id, label) = entry?;
58                labels_by_id.insert(id.value(), label.value().to_owned());
59            }
60        }
61        {
62            let table = read.open_table(marsdb_storage::tables::LABEL_TO_ID)?;
63            let mut count = 0usize;
64            for entry in table.iter()? {
65                let (label, id) = entry?;
66                count += 1;
67                if labels_by_id.get(&id.value()).map(String::as_str) != Some(label.value()) {
68                    return Err(GraphError::CorruptData(format!(
69                        "label mapping {:?} -> {} has no matching reverse mapping",
70                        label.value(),
71                        id.value()
72                    )));
73                }
74            }
75            if count != labels_by_id.len() {
76                return Err(GraphError::CorruptData(
77                    "label mapping tables have different entry counts".into(),
78                ));
79            }
80        }
81
82        let mut nodes = BTreeMap::<u64, Vec<u32>>::new();
83        {
84            let table = read.open_table(marsdb_storage::tables::NODES)?;
85            for entry in table.iter()? {
86                let (id, value) = entry?;
87                let record: NodeRecord = decode(value.value())?;
88                for label_id in &record.label_ids {
89                    if !labels_by_id.contains_key(label_id) {
90                        return Err(GraphError::CorruptData(format!(
91                            "node {} references unknown label {}",
92                            id.value(),
93                            label_id
94                        )));
95                    }
96                }
97                nodes.insert(id.value(), record.label_ids);
98            }
99        }
100
101        let mut indexed_labels = BTreeSet::new();
102        {
103            let table = read.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
104            for entry in table.iter()? {
105                let (label_id, values) = entry?;
106                let label_id = label_id.value();
107                if !labels_by_id.contains_key(&label_id) {
108                    return Err(GraphError::CorruptData(format!(
109                        "node label index references unknown label {label_id}"
110                    )));
111                }
112                for node_id in values {
113                    let node_id = node_id?.value();
114                    let Some(node_labels) = nodes.get(&node_id) else {
115                        return Err(GraphError::CorruptData(format!(
116                            "node label index references missing node {node_id}"
117                        )));
118                    };
119                    if !node_labels.contains(&label_id) {
120                        return Err(GraphError::CorruptData(format!(
121                            "node label index has label {label_id} for node {node_id}, but the node does not"
122                        )));
123                    }
124                    indexed_labels.insert((label_id, node_id));
125                }
126            }
127        }
128        for (node_id, label_ids) in &nodes {
129            for label_id in label_ids {
130                if !indexed_labels.contains(&(*label_id, *node_id)) {
131                    return Err(GraphError::CorruptData(format!(
132                        "node {node_id} has label {label_id} but is missing from the label index"
133                    )));
134                }
135            }
136        }
137
138        let mut edges = BTreeMap::<u64, (u32, u64, u64)>::new();
139        {
140            let table = read.open_table(marsdb_storage::tables::EDGES)?;
141            for entry in table.iter()? {
142                let (id, value) = entry?;
143                let record: EdgeRecord = decode(value.value())?;
144                if !labels_by_id.contains_key(&record.label_id) {
145                    return Err(GraphError::CorruptData(format!(
146                        "edge {} references unknown label {}",
147                        id.value(),
148                        record.label_id
149                    )));
150                }
151                if !nodes.contains_key(&record.src) || !nodes.contains_key(&record.dst) {
152                    return Err(GraphError::CorruptData(format!(
153                        "edge {} references missing endpoint {} -> {}",
154                        id.value(),
155                        record.src,
156                        record.dst
157                    )));
158                }
159                edges.insert(id.value(), (record.label_id, record.src, record.dst));
160            }
161        }
162
163        let outgoing =
164            Self::check_adjacency(&read, marsdb_storage::tables::ADJ_OUT, &nodes, &edges, true)?;
165        let incoming =
166            Self::check_adjacency(&read, marsdb_storage::tables::ADJ_IN, &nodes, &edges, false)?;
167        for (&edge_id, &(label_id, src, dst)) in &edges {
168            if !outgoing.contains(&(src, edge_id, dst, label_id)) {
169                return Err(GraphError::CorruptData(format!(
170                    "edge {edge_id} is missing from outgoing adjacency"
171                )));
172            }
173            if !incoming.contains(&(dst, edge_id, src, label_id)) {
174                return Err(GraphError::CorruptData(format!(
175                    "edge {edge_id} is missing from incoming adjacency"
176                )));
177            }
178        }
179
180        let meta = read.open_table(marsdb_storage::tables::META)?;
181        for (counter, maximum) in [
182            ("next_node_id", nodes.keys().next_back().copied()),
183            ("next_edge_id", edges.keys().next_back().copied()),
184        ] {
185            if let Some(maximum) = maximum {
186                let stored = meta.get(counter)?.map(|value| value.value()).unwrap_or(0);
187                if stored < maximum {
188                    return Err(GraphError::CorruptData(format!(
189                        "{counter} counter {stored} is below maximum allocated id {maximum}"
190                    )));
191                }
192            }
193        }
194
195        Ok(IntegrityReport {
196            physical_was_clean,
197            labels: labels_by_id.len() as u64,
198            nodes: nodes.len() as u64,
199            edges: edges.len() as u64,
200        })
201    }
202
203    fn check_adjacency(
204        read: &ReadTransaction,
205        definition: marsdb_storage::MultimapTableDefinition<u64, &[u8]>,
206        nodes: &BTreeMap<u64, Vec<u32>>,
207        edges: &BTreeMap<u64, (u32, u64, u64)>,
208        outgoing: bool,
209    ) -> Result<BTreeSet<(u64, u64, u64, u32)>, GraphError> {
210        let table = read.open_multimap_table(definition)?;
211        let mut found = BTreeSet::new();
212        for entry in table.iter()? {
213            let (owner, values) = entry?;
214            let owner = owner.value();
215            if !nodes.contains_key(&owner) {
216                return Err(GraphError::CorruptData(format!(
217                    "adjacency references missing owner node {owner}"
218                )));
219            }
220            for value in values {
221                let adjacency = AdjEntry::decode(value?.value())?;
222                let Some(&(label_id, src, dst)) = edges.get(&adjacency.edge_id.0) else {
223                    return Err(GraphError::CorruptData(format!(
224                        "adjacency references missing edge {}",
225                        adjacency.edge_id.0
226                    )));
227                };
228                let expected = if outgoing { (src, dst) } else { (dst, src) };
229                if owner != expected.0
230                    || adjacency.other.0 != expected.1
231                    || adjacency.label_id != label_id
232                {
233                    return Err(GraphError::CorruptData(format!(
234                        "adjacency entry for edge {} does not match the edge record",
235                        adjacency.edge_id.0
236                    )));
237                }
238                found.insert((
239                    owner,
240                    adjacency.edge_id.0,
241                    adjacency.other.0,
242                    adjacency.label_id,
243                ));
244            }
245        }
246        Ok(found)
247    }
248
249    /// Open a write transaction spanning multiple graph operations. Callers
250    /// (e.g. the query executor) drive an entire Cypher statement through
251    /// the `*_in_txn` methods below using this one transaction, then call
252    /// `write_txn.commit()` themselves — this is the crash-safety boundary
253    /// from the plan: one statement = one transaction, not one transaction
254    /// per individual node/edge write.
255    ///
256    /// v1 uses a write transaction even for pure-read statements (rather
257    /// than a separate read-only path) to keep one code path and guarantee
258    /// every statement — reads included — sees one consistent snapshot.
259    /// Trade-off: this serializes concurrent readers behind redb's
260    /// single-writer lock instead of allowing true concurrent reads; a
261    /// read-only transaction path is the natural follow-up if read
262    /// concurrency becomes a bottleneck.
263    pub fn begin_write(&self) -> Result<WriteTransaction, GraphError> {
264        Ok(self.storage.begin_write()?)
265    }
266
267    /// Open a read transaction for a statement that never mutates
268    /// anything (`MATCH ... RETURN`) — a consistent point-in-time
269    /// snapshot that runs alongside any concurrent readers or a
270    /// concurrent writer without contending for redb's single-writer
271    /// lock. No commit/abort: a read transaction has nothing to roll
272    /// back, it just releases on drop.
273    pub fn begin_read(&self) -> Result<ReadTransaction, GraphError> {
274        Ok(self.storage.begin_read()?)
275    }
276
277    /// Commit a transaction obtained from [`begin_write`](Self::begin_write).
278    pub fn commit(write_txn: WriteTransaction) -> Result<(), GraphError> {
279        write_txn.commit()?;
280        Ok(())
281    }
282
283    /// Abort (roll back) a transaction obtained from
284    /// [`begin_write`](Self::begin_write), discarding any writes made
285    /// through it.
286    pub fn abort(write_txn: WriteTransaction) -> Result<(), GraphError> {
287        write_txn.abort()?;
288        Ok(())
289    }
290
291    pub fn create_node(
292        &self,
293        labels: &[&str],
294        props: BTreeMap<String, PropertyValue>,
295    ) -> Result<NodeId, GraphError> {
296        let write_txn = self.begin_write()?;
297        let id = Self::create_node_in_txn(&write_txn, labels, props)?;
298        write_txn.commit()?;
299        Ok(id)
300    }
301
302    pub fn create_node_in_txn(
303        write_txn: &WriteTransaction,
304        labels: &[&str],
305        props: BTreeMap<String, PropertyValue>,
306    ) -> Result<NodeId, GraphError> {
307        let label_ids = labels
308            .iter()
309            .map(|l| intern_label(write_txn, l))
310            .collect::<Result<Vec<_>, _>>()?;
311        let id = next_id(write_txn, "next_node_id")?;
312        let record = NodeRecord {
313            label_ids: label_ids.clone(),
314            props,
315        };
316        let bytes = encode(&record)?;
317        let mut nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
318        nodes.insert(id, bytes.as_slice())?;
319        drop(nodes);
320        let mut label_index =
321            write_txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
322        for &label_id in &label_ids {
323            label_index.insert(label_id, id)?;
324        }
325        drop(label_index);
326        crate::index::on_node_created(write_txn, 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) => Some(decode(guard.value())?),
340                None => None,
341            };
342            found
343        };
344        let Some(record) = record else {
345            return Ok(None);
346        };
347        let labels = record
348            .label_ids
349            .iter()
350            .map(|&lid| resolve_label(txn, lid))
351            .collect::<Result<Vec<_>, _>>()?;
352        Ok(Some(Node {
353            id,
354            labels,
355            props: record.props,
356        }))
357    }
358
359    pub fn create_edge(
360        &self,
361        label: &str,
362        src: NodeId,
363        dst: NodeId,
364        props: BTreeMap<String, PropertyValue>,
365    ) -> Result<EdgeId, GraphError> {
366        let write_txn = self.begin_write()?;
367        let id = Self::create_edge_in_txn(&write_txn, label, src, dst, props)?;
368        write_txn.commit()?;
369        Ok(id)
370    }
371
372    pub fn create_edge_in_txn(
373        write_txn: &WriteTransaction,
374        label: &str,
375        src: NodeId,
376        dst: NodeId,
377        props: BTreeMap<String, PropertyValue>,
378    ) -> Result<EdgeId, GraphError> {
379        {
380            let nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
381            if nodes.get(src.0)?.is_none() {
382                return Err(GraphError::NodeNotFound(src));
383            }
384            if nodes.get(dst.0)?.is_none() {
385                return Err(GraphError::NodeNotFound(dst));
386            }
387        }
388        let label_id = intern_label(write_txn, label)?;
389        let id = next_id(write_txn, "next_edge_id")?;
390        let record = EdgeRecord {
391            label_id,
392            src: src.0,
393            dst: dst.0,
394            props,
395        };
396        let bytes = encode(&record)?;
397        let mut edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
398        edges.insert(id, bytes.as_slice())?;
399
400        let out_entry = AdjEntry {
401            edge_id: EdgeId(id),
402            other: dst,
403            label_id,
404        }
405        .encode();
406        let in_entry = AdjEntry {
407            edge_id: EdgeId(id),
408            other: src,
409            label_id,
410        }
411        .encode();
412        let mut adj_out = write_txn.open_multimap_table(marsdb_storage::tables::ADJ_OUT)?;
413        adj_out.insert(src.0, out_entry.as_slice())?;
414        let mut adj_in = write_txn.open_multimap_table(marsdb_storage::tables::ADJ_IN)?;
415        adj_in.insert(dst.0, in_entry.as_slice())?;
416        Ok(EdgeId(id))
417    }
418
419    pub fn get_edge(&self, id: EdgeId) -> Result<Option<Edge>, GraphError> {
420        let read_txn = self.begin_read()?;
421        Self::get_edge_in_txn(Txn::Read(&read_txn), id)
422    }
423
424    pub fn get_edge_in_txn(txn: Txn, id: EdgeId) -> Result<Option<Edge>, GraphError> {
425        let record: Option<EdgeRecord> = {
426            let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
427            let found = match edges.get(id.0)? {
428                Some(guard) => Some(decode(guard.value())?),
429                None => None,
430            };
431            found
432        };
433        let Some(record) = record else {
434            return Ok(None);
435        };
436        let label = resolve_label(txn, record.label_id)?;
437        Ok(Some(Edge {
438            id,
439            label,
440            src: NodeId(record.src),
441            dst: NodeId(record.dst),
442            props: record.props,
443        }))
444    }
445
446    /// Neighbors of `node` in `dir`, optionally filtered by edge label.
447    /// Reads directly from the adjacency multimap without touching `edges`.
448    pub fn neighbors(
449        &self,
450        node: NodeId,
451        dir: Direction,
452        label_filter: Option<&str>,
453    ) -> Result<Vec<AdjEntry>, GraphError> {
454        let read_txn = self.begin_read()?;
455        Self::neighbors_in_txn(Txn::Read(&read_txn), node, dir, label_filter)
456    }
457
458    pub fn neighbors_in_txn(
459        txn: Txn,
460        node: NodeId,
461        dir: Direction,
462        label_filter: Option<&str>,
463    ) -> Result<Vec<AdjEntry>, GraphError> {
464        let label_id_filter = match label_filter {
465            Some(l) => match lookup_label_id(txn, l)? {
466                Some(id) => Some(id),
467                None => return Ok(Vec::new()),
468            },
469            None => None,
470        };
471        let mut result = Vec::new();
472        let table_def = match dir {
473            Direction::Out => marsdb_storage::tables::ADJ_OUT,
474            Direction::In => marsdb_storage::tables::ADJ_IN,
475        };
476        let table = txn.open_multimap_table(table_def)?;
477        for item in table.get(node.0)? {
478            let entry = AdjEntry::decode(item?.value())?;
479            if label_id_filter.is_none_or(|lid| lid == entry.label_id) {
480                result.push(entry);
481            }
482        }
483        Ok(result)
484    }
485
486    pub fn delete_edge(&self, id: EdgeId) -> Result<bool, GraphError> {
487        let write_txn = self.begin_write()?;
488        let removed = Self::delete_edge_in_txn(&write_txn, id)?;
489        write_txn.commit()?;
490        Ok(removed)
491    }
492
493    pub fn delete_edge_in_txn(
494        write_txn: &WriteTransaction,
495        id: EdgeId,
496    ) -> Result<bool, GraphError> {
497        let record_bytes: Option<Vec<u8>> = {
498            let mut edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
499            let removed = edges.remove(id.0)?.map(|guard| guard.value().to_vec());
500            removed
501        };
502        let Some(record_bytes) = record_bytes else {
503            return Ok(false);
504        };
505        let record: EdgeRecord = decode(&record_bytes)?;
506        let out_entry = AdjEntry {
507            edge_id: id,
508            other: NodeId(record.dst),
509            label_id: record.label_id,
510        }
511        .encode();
512        let in_entry = AdjEntry {
513            edge_id: id,
514            other: NodeId(record.src),
515            label_id: record.label_id,
516        }
517        .encode();
518        {
519            let mut adj_out = write_txn.open_multimap_table(marsdb_storage::tables::ADJ_OUT)?;
520            adj_out.remove(record.src, out_entry.as_slice())?;
521        }
522        {
523            let mut adj_in = write_txn.open_multimap_table(marsdb_storage::tables::ADJ_IN)?;
524            adj_in.remove(record.dst, in_entry.as_slice())?;
525        }
526        Ok(true)
527    }
528
529    /// Delete a node. If `detach` is false and the node has incident edges,
530    /// returns `GraphError::NodeHasEdges` instead of deleting anything.
531    pub fn delete_node(&self, id: NodeId, detach: bool) -> Result<bool, GraphError> {
532        let write_txn = self.begin_write()?;
533        let existed = Self::delete_node_in_txn(&write_txn, id, detach)?;
534        write_txn.commit()?;
535        Ok(existed)
536    }
537
538    pub fn delete_node_in_txn(
539        write_txn: &WriteTransaction,
540        id: NodeId,
541        detach: bool,
542    ) -> Result<bool, GraphError> {
543        let mut incident: Vec<EdgeId> = Vec::new();
544        {
545            let adj_out = write_txn.open_multimap_table(marsdb_storage::tables::ADJ_OUT)?;
546            for item in adj_out.get(id.0)? {
547                incident.push(AdjEntry::decode(item?.value())?.edge_id);
548            }
549            let adj_in = write_txn.open_multimap_table(marsdb_storage::tables::ADJ_IN)?;
550            for item in adj_in.get(id.0)? {
551                incident.push(AdjEntry::decode(item?.value())?.edge_id);
552            }
553        }
554        if !incident.is_empty() && !detach {
555            return Err(GraphError::NodeHasEdges(id));
556        }
557        for edge_id in incident {
558            Self::delete_edge_in_txn(write_txn, edge_id)?;
559        }
560        let removed_bytes: Option<Vec<u8>> = {
561            let mut nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
562            let removed = nodes.remove(id.0)?.map(|guard| guard.value().to_vec());
563            removed
564        };
565        let Some(removed_bytes) = removed_bytes else {
566            return Ok(false);
567        };
568        let record: NodeRecord = decode(&removed_bytes)?;
569        let mut label_index =
570            write_txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
571        for &label_id in &record.label_ids {
572            label_index.remove(label_id, id.0)?;
573        }
574        drop(label_index);
575        crate::index::on_node_deleted(write_txn, id.0, &record.label_ids, &record.props)?;
576        Ok(true)
577    }
578
579    /// Declares an index on `(label, prop)`, backfilling it from every
580    /// existing node with `label` — see `index::create_index`'s own docs
581    /// for the exact semantics (idempotency, unique-violation behavior).
582    pub fn create_index(&self, label: &str, prop: &str, unique: bool) -> Result<(), GraphError> {
583        let write_txn = self.begin_write()?;
584        Self::create_index_in_txn(&write_txn, label, prop, unique)?;
585        write_txn.commit()?;
586        Ok(())
587    }
588
589    /// Same as `create_index`, but against an already-open
590    /// `WriteTransaction` — for a caller (`CREATE INDEX` as a Cypher
591    /// statement) that's already inside one transaction and must commit
592    /// or abort it as a whole, not open a second one (redb allows only one
593    /// writer at a time; opening a second would deadlock).
594    pub fn create_index_in_txn(
595        write_txn: &WriteTransaction,
596        label: &str,
597        prop: &str,
598        unique: bool,
599    ) -> Result<(), GraphError> {
600        crate::index::create_index(write_txn, label, prop, unique)
601    }
602
603    /// `None` means no index is declared on `(label, prop)`.
604    pub fn index_def(
605        &self,
606        label: &str,
607        prop: &str,
608    ) -> Result<Option<crate::IndexDef>, GraphError> {
609        let read_txn = self.begin_read()?;
610        crate::index::lookup_index_def(Txn::Read(&read_txn), label, prop)
611    }
612
613    /// Same as `index_def`, but against an already-open `Txn` — for a
614    /// caller (the query planner/executor) that's already inside one
615    /// transaction and needs a consistent view, not a fresh snapshot.
616    pub fn index_def_in_txn(
617        txn: Txn,
618        label: &str,
619        prop: &str,
620    ) -> Result<Option<crate::IndexDef>, GraphError> {
621        crate::index::lookup_index_def(txn, label, prop)
622    }
623
624    /// Same as `lookup_by_index`, but against an already-open `Txn`.
625    pub fn lookup_by_index_in_txn(
626        txn: Txn,
627        label: &str,
628        prop: &str,
629        value: &PropertyValue,
630    ) -> Result<Vec<NodeId>, GraphError> {
631        crate::index::lookup_exact(txn, label, prop, value, None)
632    }
633
634    /// Same as `lookup_by_index_in_txn`, but stops once `limit` nodes are
635    /// found — the storage-level end of `LIMIT` push-down through an
636    /// `IndexSeek` (see `marsdb_query::planner`/`executor::stream_index_seek`).
637    pub fn lookup_by_index_limited_in_txn(
638        txn: Txn,
639        label: &str,
640        prop: &str,
641        value: &PropertyValue,
642        limit: usize,
643    ) -> Result<Vec<NodeId>, GraphError> {
644        crate::index::lookup_exact(txn, label, prop, value, Some(limit))
645    }
646
647    /// Cheap, exact count of nodes under `(label, prop) = value` — for the
648    /// query planner to compare selectivity between several indexed
649    /// equality candidates, not for fetching the nodes themselves (see
650    /// `lookup_by_index_in_txn`). O(1), same contract as `lookup_by_index`
651    /// re: "no index" vs "index, no match" both reading as `0`.
652    pub fn index_match_count_in_txn(
653        txn: Txn,
654        label: &str,
655        prop: &str,
656        value: &PropertyValue,
657    ) -> Result<u64, GraphError> {
658        crate::index::match_count(txn, label, prop, value)
659    }
660
661    /// Every node currently indexed under `(label, prop) = value`. Empty
662    /// (not an error) if no such index exists — check `index_def` first if
663    /// the caller needs to distinguish "no index" from "index, no match".
664    pub fn lookup_by_index(
665        &self,
666        label: &str,
667        prop: &str,
668        value: &PropertyValue,
669    ) -> Result<Vec<NodeId>, GraphError> {
670        let read_txn = self.begin_read()?;
671        crate::index::lookup_exact(Txn::Read(&read_txn), label, prop, value, None)
672    }
673
674    pub fn set_node_prop(
675        &self,
676        id: NodeId,
677        key: &str,
678        value: PropertyValue,
679    ) -> Result<bool, GraphError> {
680        let write_txn = self.begin_write()?;
681        let updated = Self::set_node_prop_in_txn(&write_txn, id, key, value)?;
682        if updated {
683            write_txn.commit()?;
684        } else {
685            write_txn.abort()?;
686        }
687        Ok(updated)
688    }
689
690    pub fn set_node_prop_in_txn(
691        write_txn: &WriteTransaction,
692        id: NodeId,
693        key: &str,
694        value: PropertyValue,
695    ) -> Result<bool, GraphError> {
696        let bytes_opt: Option<Vec<u8>> = {
697            let nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
698            let found = nodes.get(id.0)?.map(|g| g.value().to_vec());
699            found
700        };
701        let Some(bytes) = bytes_opt else {
702            return Ok(false);
703        };
704        let mut record: NodeRecord = decode(&bytes)?;
705        let old_value = record.props.insert(key.to_string(), value.clone());
706        let new_bytes = encode(&record)?;
707        let mut nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
708        nodes.insert(id.0, new_bytes.as_slice())?;
709        drop(nodes);
710        crate::index::on_node_prop_changed(
711            write_txn,
712            id.0,
713            &record.label_ids,
714            key,
715            old_value.as_ref(),
716            Some(&value),
717        )?;
718        Ok(true)
719    }
720
721    pub fn set_edge_prop(
722        &self,
723        id: EdgeId,
724        key: &str,
725        value: PropertyValue,
726    ) -> Result<bool, GraphError> {
727        let write_txn = self.begin_write()?;
728        let updated = Self::set_edge_prop_in_txn(&write_txn, id, key, value)?;
729        if updated {
730            write_txn.commit()?;
731        } else {
732            write_txn.abort()?;
733        }
734        Ok(updated)
735    }
736
737    pub fn set_edge_prop_in_txn(
738        write_txn: &WriteTransaction,
739        id: EdgeId,
740        key: &str,
741        value: PropertyValue,
742    ) -> Result<bool, GraphError> {
743        let bytes_opt: Option<Vec<u8>> = {
744            let edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
745            let found = edges.get(id.0)?.map(|g| g.value().to_vec());
746            found
747        };
748        let Some(bytes) = bytes_opt else {
749            return Ok(false);
750        };
751        let mut record: EdgeRecord = decode(&bytes)?;
752        record.props.insert(key.to_string(), value);
753        let new_bytes = encode(&record)?;
754        let mut edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
755        edges.insert(id.0, new_bytes.as_slice())?;
756        Ok(true)
757    }
758
759    pub fn remove_node_prop_in_txn(
760        write_txn: &WriteTransaction,
761        id: NodeId,
762        key: &str,
763    ) -> Result<bool, GraphError> {
764        let bytes_opt: Option<Vec<u8>> = {
765            let nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
766            let found = nodes.get(id.0)?.map(|g| g.value().to_vec());
767            found
768        };
769        let Some(bytes) = bytes_opt else {
770            return Ok(false);
771        };
772        let mut record: NodeRecord = decode(&bytes)?;
773        let old_value = record.props.remove(key);
774        let new_bytes = encode(&record)?;
775        let mut nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
776        nodes.insert(id.0, new_bytes.as_slice())?;
777        drop(nodes);
778        crate::index::on_node_prop_changed(
779            write_txn,
780            id.0,
781            &record.label_ids,
782            key,
783            old_value.as_ref(),
784            None,
785        )?;
786        Ok(true)
787    }
788
789    pub fn remove_edge_prop_in_txn(
790        write_txn: &WriteTransaction,
791        id: EdgeId,
792        key: &str,
793    ) -> Result<bool, GraphError> {
794        let bytes_opt: Option<Vec<u8>> = {
795            let edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
796            let found = edges.get(id.0)?.map(|g| g.value().to_vec());
797            found
798        };
799        let Some(bytes) = bytes_opt else {
800            return Ok(false);
801        };
802        let mut record: EdgeRecord = decode(&bytes)?;
803        record.props.remove(key);
804        let new_bytes = encode(&record)?;
805        let mut edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
806        edges.insert(id.0, new_bytes.as_slice())?;
807        Ok(true)
808    }
809
810    /// Adds `label` to `id`'s label set -- a no-op (not an error) if it's
811    /// already there, same idempotent-add semantics real Cypher's `SET
812    /// n:Label` has.
813    pub fn add_node_label_in_txn(
814        write_txn: &WriteTransaction,
815        id: NodeId,
816        label: &str,
817    ) -> Result<bool, GraphError> {
818        let bytes_opt: Option<Vec<u8>> = {
819            let nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
820            let found = nodes.get(id.0)?.map(|g| g.value().to_vec());
821            found
822        };
823        let Some(bytes) = bytes_opt else {
824            return Ok(false);
825        };
826        let mut record: NodeRecord = decode(&bytes)?;
827        let label_id = intern_label(write_txn, label)?;
828        if !record.label_ids.contains(&label_id) {
829            record.label_ids.push(label_id);
830            let new_bytes = encode(&record)?;
831            let mut nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
832            nodes.insert(id.0, new_bytes.as_slice())?;
833            let mut label_index =
834                write_txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
835            label_index.insert(label_id, id.0)?;
836            drop(label_index);
837            crate::index::on_node_created(write_txn, id.0, &[label_id], &record.props)?;
838        }
839        Ok(true)
840    }
841
842    /// Removes `label` from `id`'s label set -- a no-op (not an error) if
843    /// it's not there (label unknown entirely, or known but not on this
844    /// node), same as real Cypher's `REMOVE n:Label`.
845    pub fn remove_node_label_in_txn(
846        write_txn: &WriteTransaction,
847        id: NodeId,
848        label: &str,
849    ) -> Result<bool, GraphError> {
850        let bytes_opt: Option<Vec<u8>> = {
851            let nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
852            let found = nodes.get(id.0)?.map(|g| g.value().to_vec());
853            found
854        };
855        let Some(bytes) = bytes_opt else {
856            return Ok(false);
857        };
858        let Some(label_id) = lookup_label_id(Txn::Write(write_txn), label)? else {
859            return Ok(true);
860        };
861        let mut record: NodeRecord = decode(&bytes)?;
862        if let Some(pos) = record.label_ids.iter().position(|&l| l == label_id) {
863            record.label_ids.remove(pos);
864            let new_bytes = encode(&record)?;
865            let mut nodes = write_txn.open_table(marsdb_storage::tables::NODES)?;
866            nodes.insert(id.0, new_bytes.as_slice())?;
867            let mut label_index =
868                write_txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
869            label_index.remove(label_id, id.0)?;
870            drop(label_index);
871            crate::index::on_node_deleted(write_txn, id.0, &[label_id], &record.props)?;
872        }
873        Ok(true)
874    }
875
876    /// Full scan of all nodes, optionally filtered by label. v1 has no
877    /// secondary index on label, so this is a linear scan of the table.
878    pub fn all_nodes(&self, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
879        let read_txn = self.begin_read()?;
880        Self::all_nodes_in_txn(Txn::Read(&read_txn), label_filter)
881    }
882
883    /// Scan only graph identities, without decoding node records. Query
884    /// pipelines use this to defer record/property loading until a filter or
885    /// projection actually needs it.
886    pub fn all_node_ids_limited_in_txn(
887        txn: Txn,
888        label_filter: Option<&str>,
889        limit: usize,
890    ) -> Result<Vec<NodeId>, GraphError> {
891        let Some(label_filter) = label_filter else {
892            let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
893            return nodes
894                .iter()?
895                .take(limit)
896                .map(|entry| {
897                    entry
898                        .map(|(key, _)| NodeId(key.value()))
899                        .map_err(Into::into)
900                })
901                .collect();
902        };
903        let Some(label_id) = lookup_label_id(txn, label_filter)? else {
904            return Ok(Vec::new());
905        };
906        let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
907        let ids = label_index
908            .get(label_id)?
909            .take(limit)
910            .map(|entry| entry.map(|value| NodeId(value.value())).map_err(Into::into))
911            .collect::<Result<Vec<_>, GraphError>>()?;
912        drop(label_index);
913        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
914        for id in &ids {
915            if nodes.get(id.0)?.is_none() {
916                return Err(GraphError::CorruptData(format!(
917                    "node label index references missing node {}",
918                    id.0
919                )));
920            }
921        }
922        Ok(ids)
923    }
924
925    pub fn all_nodes_in_txn(txn: Txn, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
926        Self::all_nodes_limited_in_txn(txn, label_filter, usize::MAX)
927    }
928
929    /// Same as `all_nodes_in_txn`, but stops once `limit` nodes are found --
930    /// the storage-level end of `LIMIT` push-down (see the executor's
931    /// `scan()`/`eval_plan` docs for the query-level half): a query whose
932    /// entire plan is a bare scan feeding straight into a `LIMIT` doesn't
933    /// need to touch rows past the first `limit`, whether or not a label
934    /// filter narrows it first.
935    pub fn all_nodes_limited_in_txn(
936        txn: Txn,
937        label_filter: Option<&str>,
938        limit: usize,
939    ) -> Result<Vec<Node>, GraphError> {
940        // A label filter goes through NODE_LABEL_INDEX (label_id -> node_ids)
941        // plus a point lookup per match, instead of scanning every row in
942        // NODES — cost scales with the number of matching rows, not the
943        // table size. No filter means every row is wanted anyway, so a full
944        // scan is already optimal; the index wouldn't help.
945        let Some(label_filter) = label_filter else {
946            let mut result = Vec::new();
947            let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
948            for item in nodes.iter()? {
949                if result.len() >= limit {
950                    break;
951                }
952                let (key, value) = item?;
953                let record: NodeRecord = decode(value.value())?;
954                let labels = record
955                    .label_ids
956                    .iter()
957                    .map(|&lid| resolve_label(txn, lid))
958                    .collect::<Result<Vec<_>, _>>()?;
959                result.push(Node {
960                    id: NodeId(key.value()),
961                    labels,
962                    props: record.props,
963                });
964            }
965            return Ok(result);
966        };
967        let Some(label_id) = lookup_label_id(txn, label_filter)? else {
968            return Ok(Vec::new());
969        };
970        let node_ids: Vec<u64> = {
971            let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
972            // `.take(limit)` here, not a `.truncate()` after collecting --
973            // stops walking the multimap's own entries past `limit`, not
974            // just the (more expensive) per-id NODES point-reads below.
975            // Measured difference: without this, a labeled LIMIT query's
976            // cost still scaled with the *matching* row count, not `limit`
977            // (see BENCHMARKS.md's `execute_scan_limit_pushdown` numbers).
978            let ids: Vec<u64> = label_index
979                .get(label_id)?
980                .take(limit)
981                .map(|item| item.map(|g| g.value()))
982                .collect::<Result<_, _>>()?;
983            ids
984        };
985        let mut result = Vec::with_capacity(node_ids.len());
986        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
987        for id in node_ids {
988            let guard = nodes.get(id)?.ok_or_else(|| {
989                GraphError::CorruptData(format!("node label index references missing node {}", id))
990            })?;
991            let record: NodeRecord = decode(guard.value())?;
992            drop(guard);
993            let labels = record
994                .label_ids
995                .iter()
996                .map(|&lid| resolve_label(txn, lid))
997                .collect::<Result<Vec<_>, _>>()?;
998            result.push(Node {
999                id: NodeId(id),
1000                labels,
1001                props: record.props,
1002            });
1003        }
1004        Ok(result)
1005    }
1006}
1007
1008#[cfg(test)]
1009mod tests {
1010    use super::*;
1011
1012    #[test]
1013    fn integrity_check_rejects_missing_node_label_index_entry() {
1014        let mut store = GraphStore::open_memory().unwrap();
1015        let node = store.create_node(&["Person"], BTreeMap::new()).unwrap();
1016
1017        let write = store.begin_write().unwrap();
1018        let label_id = {
1019            let labels = write
1020                .open_table(marsdb_storage::tables::LABEL_TO_ID)
1021                .unwrap();
1022            let id = labels.get("Person").unwrap().unwrap().value();
1023            id
1024        };
1025        write
1026            .open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)
1027            .unwrap()
1028            .remove(label_id, node.0)
1029            .unwrap();
1030        write.commit().unwrap();
1031
1032        let error = store.check_integrity().unwrap_err();
1033        assert!(
1034            matches!(error, GraphError::CorruptData(message) if message.contains("missing from the label index"))
1035        );
1036    }
1037
1038    #[test]
1039    fn integrity_check_rejects_dangling_adjacency_entry() {
1040        let mut store = GraphStore::open_memory().unwrap();
1041        let node = store.create_node(&[], BTreeMap::new()).unwrap();
1042
1043        let write = store.begin_write().unwrap();
1044        let bytes = AdjEntry {
1045            edge_id: EdgeId(999),
1046            other: node,
1047            label_id: 0,
1048        }
1049        .encode();
1050        write
1051            .open_multimap_table(marsdb_storage::tables::ADJ_OUT)
1052            .unwrap()
1053            .insert(node.0, bytes.as_slice())
1054            .unwrap();
1055        write.commit().unwrap();
1056
1057        let error = store.check_integrity().unwrap_err();
1058        assert!(
1059            matches!(error, GraphError::CorruptData(message) if message.contains("missing edge 999"))
1060        );
1061    }
1062
1063    #[test]
1064    fn create_index_backfills_existing_nodes() {
1065        let store = GraphStore::open_memory().unwrap();
1066        let mut alice_props = BTreeMap::new();
1067        alice_props.insert(
1068            "email".to_string(),
1069            PropertyValue::String("alice@x.com".to_string()),
1070        );
1071        let alice = store.create_node(&["Person"], alice_props).unwrap();
1072        let mut bob_props = BTreeMap::new();
1073        bob_props.insert(
1074            "email".to_string(),
1075            PropertyValue::String("bob@x.com".to_string()),
1076        );
1077        store.create_node(&["Person"], bob_props).unwrap();
1078        // A Person with no email at all -- must not show up under any lookup.
1079        store.create_node(&["Person"], BTreeMap::new()).unwrap();
1080
1081        store.create_index("Person", "email", false).unwrap();
1082
1083        let found = store
1084            .lookup_by_index(
1085                "Person",
1086                "email",
1087                &PropertyValue::String("alice@x.com".to_string()),
1088            )
1089            .unwrap();
1090        assert_eq!(found, vec![alice]);
1091    }
1092
1093    #[test]
1094    fn create_index_rejects_duplicate_unique_value() {
1095        let store = GraphStore::open_memory().unwrap();
1096        let mut props1 = BTreeMap::new();
1097        props1.insert(
1098            "email".to_string(),
1099            PropertyValue::String("same@x.com".to_string()),
1100        );
1101        store.create_node(&["Person"], props1).unwrap();
1102        let mut props2 = BTreeMap::new();
1103        props2.insert(
1104            "email".to_string(),
1105            PropertyValue::String("same@x.com".to_string()),
1106        );
1107        store.create_node(&["Person"], props2).unwrap();
1108
1109        let error = store.create_index("Person", "email", true).unwrap_err();
1110        assert!(matches!(
1111            error,
1112            GraphError::UniqueConstraintViolation { .. }
1113        ));
1114
1115        // A rejected unique index must not partially exist.
1116        assert!(store.index_def("Person", "email").unwrap().is_none());
1117    }
1118
1119    #[test]
1120    fn lookup_by_index_on_undeclared_index_is_empty_not_an_error() {
1121        let store = GraphStore::open_memory().unwrap();
1122        store.create_node(&["Person"], BTreeMap::new()).unwrap();
1123        let found = store
1124            .lookup_by_index("Person", "email", &PropertyValue::String("x".to_string()))
1125            .unwrap();
1126        assert_eq!(found, Vec::new());
1127        assert!(store.index_def("Person", "email").unwrap().is_none());
1128    }
1129
1130    #[test]
1131    fn index_survives_reopen() {
1132        let dir = tempfile::tempdir().unwrap();
1133        let path = dir.path().join("index.db");
1134        {
1135            let store = GraphStore::open_file(&path).unwrap();
1136            let mut props = BTreeMap::new();
1137            props.insert(
1138                "email".to_string(),
1139                PropertyValue::String("x@x.com".to_string()),
1140            );
1141            store.create_node(&["Person"], props).unwrap();
1142            store.create_index("Person", "email", false).unwrap();
1143        }
1144        let store = GraphStore::open_file(&path).unwrap();
1145        assert!(store.index_def("Person", "email").unwrap().is_some());
1146        let found = store
1147            .lookup_by_index(
1148                "Person",
1149                "email",
1150                &PropertyValue::String("x@x.com".to_string()),
1151            )
1152            .unwrap();
1153        assert_eq!(found.len(), 1);
1154    }
1155
1156    #[test]
1157    fn create_node_after_index_declared_is_indexed_immediately() {
1158        let store = GraphStore::open_memory().unwrap();
1159        store.create_index("Person", "email", false).unwrap();
1160        let mut props = BTreeMap::new();
1161        props.insert(
1162            "email".to_string(),
1163            PropertyValue::String("new@x.com".to_string()),
1164        );
1165        let node = store.create_node(&["Person"], props).unwrap();
1166
1167        let found = store
1168            .lookup_by_index(
1169                "Person",
1170                "email",
1171                &PropertyValue::String("new@x.com".to_string()),
1172            )
1173            .unwrap();
1174        assert_eq!(found, vec![node]);
1175    }
1176
1177    #[test]
1178    fn set_node_prop_moves_the_index_entry() {
1179        let store = GraphStore::open_memory().unwrap();
1180        let mut props = BTreeMap::new();
1181        props.insert(
1182            "email".to_string(),
1183            PropertyValue::String("old@x.com".to_string()),
1184        );
1185        let node = store.create_node(&["Person"], props).unwrap();
1186        store.create_index("Person", "email", false).unwrap();
1187
1188        store
1189            .set_node_prop(
1190                node,
1191                "email",
1192                PropertyValue::String("new@x.com".to_string()),
1193            )
1194            .unwrap();
1195
1196        assert!(store
1197            .lookup_by_index(
1198                "Person",
1199                "email",
1200                &PropertyValue::String("old@x.com".to_string())
1201            )
1202            .unwrap()
1203            .is_empty());
1204        assert_eq!(
1205            store
1206                .lookup_by_index(
1207                    "Person",
1208                    "email",
1209                    &PropertyValue::String("new@x.com".to_string())
1210                )
1211                .unwrap(),
1212            vec![node]
1213        );
1214    }
1215
1216    #[test]
1217    fn set_node_prop_enforces_unique_index() {
1218        let store = GraphStore::open_memory().unwrap();
1219        let mut props1 = BTreeMap::new();
1220        props1.insert(
1221            "email".to_string(),
1222            PropertyValue::String("a@x.com".to_string()),
1223        );
1224        store.create_node(&["Person"], props1).unwrap();
1225        let mut props2 = BTreeMap::new();
1226        props2.insert(
1227            "email".to_string(),
1228            PropertyValue::String("b@x.com".to_string()),
1229        );
1230        let node2 = store.create_node(&["Person"], props2).unwrap();
1231        store.create_index("Person", "email", true).unwrap();
1232
1233        let error = store
1234            .set_node_prop(node2, "email", PropertyValue::String("a@x.com".to_string()))
1235            .unwrap_err();
1236        assert!(matches!(
1237            error,
1238            GraphError::UniqueConstraintViolation { .. }
1239        ));
1240    }
1241
1242    #[test]
1243    fn remove_node_prop_removes_the_index_entry() {
1244        let store = GraphStore::open_memory().unwrap();
1245        let mut props = BTreeMap::new();
1246        props.insert(
1247            "email".to_string(),
1248            PropertyValue::String("gone@x.com".to_string()),
1249        );
1250        let node = store.create_node(&["Person"], props).unwrap();
1251        store.create_index("Person", "email", false).unwrap();
1252
1253        let write = store.begin_write().unwrap();
1254        GraphStore::remove_node_prop_in_txn(&write, node, "email").unwrap();
1255        write.commit().unwrap();
1256
1257        assert!(store
1258            .lookup_by_index(
1259                "Person",
1260                "email",
1261                &PropertyValue::String("gone@x.com".to_string())
1262            )
1263            .unwrap()
1264            .is_empty());
1265    }
1266
1267    #[test]
1268    fn delete_node_removes_its_index_entries() {
1269        let store = GraphStore::open_memory().unwrap();
1270        let mut props = BTreeMap::new();
1271        props.insert(
1272            "email".to_string(),
1273            PropertyValue::String("deleted@x.com".to_string()),
1274        );
1275        let node = store.create_node(&["Person"], props).unwrap();
1276        store.create_index("Person", "email", false).unwrap();
1277
1278        store.delete_node(node, false).unwrap();
1279
1280        assert!(store
1281            .lookup_by_index(
1282                "Person",
1283                "email",
1284                &PropertyValue::String("deleted@x.com".to_string())
1285            )
1286            .unwrap()
1287            .is_empty());
1288    }
1289
1290    #[test]
1291    fn add_node_label_indexes_existing_props_under_the_new_label() {
1292        let store = GraphStore::open_memory().unwrap();
1293        let mut props = BTreeMap::new();
1294        props.insert(
1295            "email".to_string(),
1296            PropertyValue::String("multi@x.com".to_string()),
1297        );
1298        let node = store.create_node(&["Contact"], props).unwrap();
1299        store.create_index("Person", "email", false).unwrap();
1300
1301        // Not indexed yet -- the node isn't a Person.
1302        assert!(store
1303            .lookup_by_index(
1304                "Person",
1305                "email",
1306                &PropertyValue::String("multi@x.com".to_string())
1307            )
1308            .unwrap()
1309            .is_empty());
1310
1311        let write = store.begin_write().unwrap();
1312        GraphStore::add_node_label_in_txn(&write, node, "Person").unwrap();
1313        write.commit().unwrap();
1314
1315        assert_eq!(
1316            store
1317                .lookup_by_index(
1318                    "Person",
1319                    "email",
1320                    &PropertyValue::String("multi@x.com".to_string())
1321                )
1322                .unwrap(),
1323            vec![node]
1324        );
1325    }
1326
1327    #[test]
1328    fn remove_node_label_removes_index_entries_under_that_label() {
1329        let store = GraphStore::open_memory().unwrap();
1330        let mut props = BTreeMap::new();
1331        props.insert(
1332            "email".to_string(),
1333            PropertyValue::String("dual@x.com".to_string()),
1334        );
1335        let node = store.create_node(&["Person", "Contact"], props).unwrap();
1336        store.create_index("Person", "email", false).unwrap();
1337        assert_eq!(
1338            store
1339                .lookup_by_index(
1340                    "Person",
1341                    "email",
1342                    &PropertyValue::String("dual@x.com".to_string())
1343                )
1344                .unwrap(),
1345            vec![node]
1346        );
1347
1348        let write = store.begin_write().unwrap();
1349        GraphStore::remove_node_label_in_txn(&write, node, "Person").unwrap();
1350        write.commit().unwrap();
1351
1352        assert!(store
1353            .lookup_by_index(
1354                "Person",
1355                "email",
1356                &PropertyValue::String("dual@x.com".to_string())
1357            )
1358            .unwrap()
1359            .is_empty());
1360    }
1361}
marsdb_graph/store.rs

marsdb_graph/
store.rs
