Skip to main content

marsdb_graph/
store.rs

1use std::collections::{BTreeMap, BTreeSet, HashMap};
2use std::path::Path;
3
4use marsdb_storage::{
5    ReadTransaction, ReadableMultimapTable, ReadableTable, ReadableTableMetadata, StorageEngine,
6    Txn, WriteTransaction,
7};
8
9use crate::encode::{
10    decode_edge, decode_node, edge_header, encode_edge, encode_node, node_label_ids, EdgeRecord,
11    NodeRecord,
12};
13use crate::error::GraphError;
14use crate::id::next_id;
15use crate::labels::{intern_label, lookup_label_id, resolve_label};
16use crate::model::{AdjEntry, Direction, Edge, EdgeId, Node, NodeId, PropertyValue};
17use crate::props::{intern_prop, prop_resolver};
18use crate::write_ctx::WriteCtx;
19
20pub struct GraphStore {
21    storage: StorageEngine,
22}
23
24/// Successful physical and logical integrity-check summary.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct IntegrityReport {
27    /// `false` means redb detected physical damage and repaired it before
28    /// MarsDB's logical checks ran.
29    pub physical_was_clean: bool,
30    pub labels: u64,
31    pub nodes: u64,
32    pub edges: u64,
33}
34
35impl GraphStore {
36    pub fn open_file(path: impl AsRef<Path>) -> Result<Self, GraphError> {
37        let store = Self {
38            storage: StorageEngine::open_file(path)?,
39        };
40        store.backfill_rel_type_counts()?;
41        Ok(store)
42    }
43
44    pub fn open_memory() -> Result<Self, GraphError> {
45        Ok(Self {
46            storage: StorageEngine::open_memory()?,
47        })
48    }
49
50    /// One-time `REL_TYPE_COUNTS` rebuild for a file written by a build
51    /// that predates the table: counts empty while `EDGES` isn't can only
52    /// mean the maintaining writes never ran, so scan every edge header
53    /// once (no property decode) and commit the tallies. A fresh or
54    /// up-to-date file exits on the first check without writing anything.
55    /// A file this build writes and an *older* build later mutates would
56    /// go stale with no way to detect it here -- tolerable by
57    /// construction, since the table is a planner statistic that can cost
58    /// a suboptimal plan but never a wrong result (see its definition).
59    fn backfill_rel_type_counts(&self) -> Result<(), GraphError> {
60        let write_txn = self.begin_write()?;
61        let up_to_date = {
62            let counts = write_txn.open_table(marsdb_storage::tables::REL_TYPE_COUNTS)?;
63            let edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
64            !counts.is_empty()? || edges.is_empty()?
65        };
66        if up_to_date {
67            write_txn.abort()?;
68            return Ok(());
69        }
70        let mut tallies: std::collections::HashMap<u32, u64> = std::collections::HashMap::new();
71        {
72            let edges = write_txn.open_table(marsdb_storage::tables::EDGES)?;
73            for entry in edges.iter()? {
74                let (_, value) = entry?;
75                let (label_id, _, _) = edge_header(value.value())?;
76                *tallies.entry(label_id).or_insert(0) += 1;
77            }
78        }
79        {
80            let mut counts = write_txn.open_table(marsdb_storage::tables::REL_TYPE_COUNTS)?;
81            for (label_id, count) in tallies {
82                counts.insert(label_id, count)?;
83            }
84        }
85        write_txn.commit()?;
86        Ok(())
87    }
88
89    pub fn backup_to(&self, path: impl AsRef<Path>) -> Result<(), GraphError> {
90        self.storage.backup_to(path)?;
91        Ok(())
92    }
93
94    /// Check physical storage plus MarsDB's graph invariants. This requires
95    /// exclusive mutable access because redb may repair physical metadata.
96    pub fn check_integrity(&mut self) -> Result<IntegrityReport, GraphError> {
97        let physical_was_clean = self.storage.check_integrity()?;
98        let read = self.storage.begin_read()?;
99
100        let mut labels_by_id = BTreeMap::new();
101        {
102            let table = read.open_table(marsdb_storage::tables::ID_TO_LABEL)?;
103            for entry in table.iter()? {
104                let (id, label) = entry?;
105                labels_by_id.insert(id.value(), label.value().to_owned());
106            }
107        }
108        {
109            let table = read.open_table(marsdb_storage::tables::LABEL_TO_ID)?;
110            let mut count = 0usize;
111            for entry in table.iter()? {
112                let (label, id) = entry?;
113                count += 1;
114                if labels_by_id.get(&id.value()).map(String::as_str) != Some(label.value()) {
115                    return Err(GraphError::CorruptData(format!(
116                        "label mapping {:?} -> {} has no matching reverse mapping",
117                        label.value(),
118                        id.value()
119                    )));
120                }
121            }
122            if count != labels_by_id.len() {
123                return Err(GraphError::CorruptData(
124                    "label mapping tables have different entry counts".into(),
125                ));
126            }
127        }
128
129        let mut nodes = BTreeMap::<u64, Vec<u32>>::new();
130        {
131            let table = read.open_table(marsdb_storage::tables::NODES)?;
132            for entry in table.iter()? {
133                let (id, value) = entry?;
134                // Header-only read -- integrity checks node labels here,
135                // never properties, so the directory stays untouched.
136                let label_ids = node_label_ids(value.value())?;
137                for label_id in &label_ids {
138                    if !labels_by_id.contains_key(label_id) {
139                        return Err(GraphError::CorruptData(format!(
140                            "node {} references unknown label {}",
141                            id.value(),
142                            label_id
143                        )));
144                    }
145                }
146                nodes.insert(id.value(), label_ids);
147            }
148        }
149
150        let mut indexed_labels = BTreeSet::new();
151        {
152            let table = read.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
153            for entry in table.iter()? {
154                let (label_id, values) = entry?;
155                let label_id = label_id.value();
156                if !labels_by_id.contains_key(&label_id) {
157                    return Err(GraphError::CorruptData(format!(
158                        "node label index references unknown label {label_id}"
159                    )));
160                }
161                for node_id in values {
162                    let node_id = node_id?.value();
163                    let Some(node_labels) = nodes.get(&node_id) else {
164                        return Err(GraphError::CorruptData(format!(
165                            "node label index references missing node {node_id}"
166                        )));
167                    };
168                    if !node_labels.contains(&label_id) {
169                        return Err(GraphError::CorruptData(format!(
170                            "node label index has label {label_id} for node {node_id}, but the node does not"
171                        )));
172                    }
173                    indexed_labels.insert((label_id, node_id));
174                }
175            }
176        }
177        for (node_id, label_ids) in &nodes {
178            for label_id in label_ids {
179                if !indexed_labels.contains(&(*label_id, *node_id)) {
180                    return Err(GraphError::CorruptData(format!(
181                        "node {node_id} has label {label_id} but is missing from the label index"
182                    )));
183                }
184            }
185        }
186
187        let mut edges = BTreeMap::<u64, (u32, u64, u64)>::new();
188        {
189            let table = read.open_table(marsdb_storage::tables::EDGES)?;
190            for entry in table.iter()? {
191                let (id, value) = entry?;
192                // Header-only, same reasoning as the node loop above.
193                let (label_id, src, dst) = edge_header(value.value())?;
194                if !labels_by_id.contains_key(&label_id) {
195                    return Err(GraphError::CorruptData(format!(
196                        "edge {} references unknown label {}",
197                        id.value(),
198                        label_id
199                    )));
200                }
201                if !nodes.contains_key(&src) || !nodes.contains_key(&dst) {
202                    return Err(GraphError::CorruptData(format!(
203                        "edge {} references missing endpoint {} -> {}",
204                        id.value(),
205                        src,
206                        dst
207                    )));
208                }
209                edges.insert(id.value(), (label_id, src, dst));
210            }
211        }
212
213        let outgoing =
214            Self::check_adjacency(&read, marsdb_storage::tables::ADJ_OUT, &nodes, &edges, true)?;
215        let incoming =
216            Self::check_adjacency(&read, marsdb_storage::tables::ADJ_IN, &nodes, &edges, false)?;
217        for (&edge_id, &(label_id, src, dst)) in &edges {
218            if !outgoing.contains(&(src, edge_id, dst, label_id)) {
219                return Err(GraphError::CorruptData(format!(
220                    "edge {edge_id} is missing from outgoing adjacency"
221                )));
222            }
223            if !incoming.contains(&(dst, edge_id, src, label_id)) {
224                return Err(GraphError::CorruptData(format!(
225                    "edge {edge_id} is missing from incoming adjacency"
226                )));
227            }
228        }
229
230        let meta = read.open_table(marsdb_storage::tables::META)?;
231        for (counter, maximum) in [
232            ("next_node_id", nodes.keys().next_back().copied()),
233            ("next_edge_id", edges.keys().next_back().copied()),
234        ] {
235            if let Some(maximum) = maximum {
236                let stored = meta.get(counter)?.map(|value| value.value()).unwrap_or(0);
237                if stored < maximum {
238                    return Err(GraphError::CorruptData(format!(
239                        "{counter} counter {stored} is below maximum allocated id {maximum}"
240                    )));
241                }
242            }
243        }
244
245        Ok(IntegrityReport {
246            physical_was_clean,
247            labels: labels_by_id.len() as u64,
248            nodes: nodes.len() as u64,
249            edges: edges.len() as u64,
250        })
251    }
252
253    fn check_adjacency(
254        read: &ReadTransaction,
255        definition: marsdb_storage::TableDefinition<(u64, u32, u64), u64>,
256        nodes: &BTreeMap<u64, Vec<u32>>,
257        edges: &BTreeMap<u64, (u32, u64, u64)>,
258        outgoing: bool,
259    ) -> Result<BTreeSet<(u64, u64, u64, u32)>, GraphError> {
260        let table = read.open_table(definition)?;
261        let mut found = BTreeSet::new();
262        for entry in table.iter()? {
263            let (key, value) = entry?;
264            let (owner, key_label_id, edge_id) = key.value();
265            let other = value.value();
266            if !nodes.contains_key(&owner) {
267                return Err(GraphError::CorruptData(format!(
268                    "adjacency references missing owner node {owner}"
269                )));
270            }
271            let Some(&(label_id, src, dst)) = edges.get(&edge_id) else {
272                return Err(GraphError::CorruptData(format!(
273                    "adjacency references missing edge {edge_id}"
274                )));
275            };
276            let expected = if outgoing { (src, dst) } else { (dst, src) };
277            if owner != expected.0 || other != expected.1 || key_label_id != label_id {
278                return Err(GraphError::CorruptData(format!(
279                    "adjacency entry for edge {edge_id} does not match the edge record"
280                )));
281            }
282            found.insert((owner, edge_id, other, key_label_id));
283        }
284        Ok(found)
285    }
286
287    /// Open a write transaction spanning multiple graph operations. Callers
288    /// (e.g. the query executor) drive an entire Cypher statement through
289    /// the `*_in_txn` methods below using this one transaction, then call
290    /// `write_txn.commit()` themselves — this is the crash-safety boundary
291    /// from the plan: one statement = one transaction, not one transaction
292    /// per individual node/edge write.
293    ///
294    /// v1 uses a write transaction even for pure-read statements (rather
295    /// than a separate read-only path) to keep one code path and guarantee
296    /// every statement — reads included — sees one consistent snapshot.
297    /// Trade-off: this serializes concurrent readers behind redb's
298    /// single-writer lock instead of allowing true concurrent reads; a
299    /// read-only transaction path is the natural follow-up if read
300    /// concurrency becomes a bottleneck.
301    pub fn begin_write(&self) -> Result<WriteTransaction, GraphError> {
302        Ok(self.storage.begin_write()?)
303    }
304
305    /// Open a read transaction for a statement that never mutates
306    /// anything (`MATCH ... RETURN`) — a consistent point-in-time
307    /// snapshot that runs alongside any concurrent readers or a
308    /// concurrent writer without contending for redb's single-writer
309    /// lock. No commit/abort: a read transaction has nothing to roll
310    /// back, it just releases on drop.
311    pub fn begin_read(&self) -> Result<ReadTransaction, GraphError> {
312        Ok(self.storage.begin_read()?)
313    }
314
315    /// Commit a transaction obtained from [`begin_write`](Self::begin_write).
316    pub fn commit(write_txn: WriteTransaction) -> Result<(), GraphError> {
317        write_txn.commit()?;
318        Ok(())
319    }
320
321    /// Abort (roll back) a transaction obtained from
322    /// [`begin_write`](Self::begin_write), discarding any writes made
323    /// through it.
324    pub fn abort(write_txn: WriteTransaction) -> Result<(), GraphError> {
325        write_txn.abort()?;
326        Ok(())
327    }
328
329    pub fn create_node(
330        &self,
331        labels: &[&str],
332        props: BTreeMap<String, PropertyValue>,
333    ) -> Result<NodeId, GraphError> {
334        let write_txn = self.begin_write()?;
335        let id = Self::create_node_in_txn(&write_txn, labels, props)?;
336        write_txn.commit()?;
337        Ok(id)
338    }
339
340    pub fn create_node_in_txn(
341        write_txn: &WriteTransaction,
342        labels: &[&str],
343        props: BTreeMap<String, PropertyValue>,
344    ) -> Result<NodeId, GraphError> {
345        let mut ctx = WriteCtx::open(write_txn);
346        Self::create_node_ctx(&mut ctx, labels, props)
347    }
348
349    fn create_node_ctx(
350        ctx: &mut WriteCtx,
351        labels: &[&str],
352        props: BTreeMap<String, PropertyValue>,
353    ) -> Result<NodeId, GraphError> {
354        let label_ids = labels
355            .iter()
356            .map(|l| intern_label(ctx, l))
357            .collect::<Result<Vec<_>, _>>()?;
358        let id = next_id(ctx, "next_node_id")?;
359        let record = NodeRecord {
360            label_ids: label_ids.clone(),
361            props,
362        };
363        let bytes = encode_node(&record, |name| intern_prop(ctx, name))?;
364        ctx.nodes()?.insert(id, bytes.as_slice())?;
365        for &label_id in &label_ids {
366            ctx.node_label_index()?.insert(label_id, id)?;
367        }
368        crate::index::on_node_created(ctx, id, &label_ids, &record.props)?;
369        Ok(NodeId(id))
370    }
371
372    pub fn get_node(&self, id: NodeId) -> Result<Option<Node>, GraphError> {
373        let read_txn = self.begin_read()?;
374        Self::get_node_in_txn(Txn::Read(&read_txn), id)
375    }
376
377    pub fn get_node_in_txn(txn: Txn, id: NodeId) -> Result<Option<Node>, GraphError> {
378        let record: Option<NodeRecord> = {
379            let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
380            let found = match nodes.get(id.0)? {
381                Some(guard) => {
382                    let mut resolve = prop_resolver(txn)?;
383                    Some(decode_node(guard.value(), &mut resolve)?)
384                }
385                None => None,
386            };
387            found
388        };
389        let Some(record) = record else {
390            return Ok(None);
391        };
392        let labels = record
393            .label_ids
394            .iter()
395            .map(|&lid| resolve_label(txn, lid))
396            .collect::<Result<Vec<_>, _>>()?;
397        Ok(Some(Node {
398            id,
399            labels,
400            props: record.props,
401        }))
402    }
403
404    /// The id interned for a property name, if any -- `None` means the
405    /// name has never been written anywhere, so no record can hold it.
406    /// Exposed for the query layer's per-property read path: names resolve
407    /// to ids once per statement there, then every row access goes through
408    /// `get_node_prop_in_txn`/`get_edge_prop_in_txn` by id.
409    pub fn lookup_prop_id_in_txn(txn: Txn, prop: &str) -> Result<Option<u32>, GraphError> {
410        crate::props::lookup_prop_id(txn, prop)
411    }
412
413    /// One property of one node, by interned prop id, without decoding the
414    /// rest of the record or resolving any names — a directory binary
415    /// search plus one value decode (the v2 read fast path; the codec
416    /// mechanism measured 79x over whole-record decode at 1-of-20 props).
417    ///
418    /// Nested `Option` distinguishes the two kinds of missing the executor
419    /// must not collapse (`lookup_prop`'s own docs): outer `None` = the
420    /// node record doesn't exist (deleted-entity error at the call site),
421    /// inner `None` = node exists, property absent (legal null).
422    pub fn get_node_prop_in_txn(
423        txn: Txn,
424        id: NodeId,
425        prop_id: u32,
426    ) -> Result<Option<Option<PropertyValue>>, GraphError> {
427        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
428        let Some(guard) = nodes.get(id.0)? else {
429            return Ok(None);
430        };
431        match crate::encode::node_prop_raw(guard.value(), prop_id)? {
432            Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
433            None => Ok(Some(None)),
434        }
435    }
436
437    /// Edge counterpart of `get_node_prop_in_txn`, same nested-`Option`
438    /// contract.
439    pub fn get_edge_prop_in_txn(
440        txn: Txn,
441        id: EdgeId,
442        prop_id: u32,
443    ) -> Result<Option<Option<PropertyValue>>, GraphError> {
444        let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
445        let Some(guard) = edges.get(id.0)? else {
446            return Ok(None);
447        };
448        match crate::encode::edge_prop_raw(guard.value(), prop_id)? {
449            Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
450            None => Ok(Some(None)),
451        }
452    }
453
454    /// Per-property reader over ONE pre-opened `NODES` handle -- for a
455    /// caller probing many nodes' properties in a loop, where
456    /// `get_node_prop_in_txn`'s per-call table open would dominate (the
457    /// mars-3va lesson: opens measured 23.67% of a bulk load). Same
458    /// nested-`Option` contract as `get_node_prop_in_txn`.
459    #[allow(clippy::type_complexity)] // the nested Option IS the contract (see get_node_prop_in_txn)
460    pub fn node_prop_reader(
461        txn: Txn<'_>,
462    ) -> Result<
463        impl FnMut(NodeId, u32) -> Result<Option<Option<PropertyValue>>, GraphError> + '_,
464        GraphError,
465    > {
466        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
467        Ok(move |id: NodeId, prop_id: u32| {
468            let Some(guard) = nodes.get(id.0)? else {
469                return Ok(None);
470            };
471            match crate::encode::node_prop_raw(guard.value(), prop_id)? {
472                Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
473                None => Ok(Some(None)),
474            }
475        })
476    }
477
478    /// Record-existence check without any decoding — for the per-property
479    /// read path when the property name was never interned (the value is
480    /// necessarily absent on every record, but a *deleted* node must still
481    /// error, not read as null).
482    pub fn node_exists_in_txn(txn: Txn, id: NodeId) -> Result<bool, GraphError> {
483        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
484        let exists = nodes.get(id.0)?.is_some();
485        Ok(exists)
486    }
487
488    /// Edge counterpart of `node_exists_in_txn`.
489    pub fn edge_exists_in_txn(txn: Txn, id: EdgeId) -> Result<bool, GraphError> {
490        let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
491        let exists = edges.get(id.0)?.is_some();
492        Ok(exists)
493    }
494
495    pub fn create_edge(
496        &self,
497        label: &str,
498        src: NodeId,
499        dst: NodeId,
500        props: BTreeMap<String, PropertyValue>,
501    ) -> Result<EdgeId, GraphError> {
502        let write_txn = self.begin_write()?;
503        let id = Self::create_edge_in_txn(&write_txn, label, src, dst, props)?;
504        write_txn.commit()?;
505        Ok(id)
506    }
507
508    pub fn create_edge_in_txn(
509        write_txn: &WriteTransaction,
510        label: &str,
511        src: NodeId,
512        dst: NodeId,
513        props: BTreeMap<String, PropertyValue>,
514    ) -> Result<EdgeId, GraphError> {
515        let mut ctx = WriteCtx::open(write_txn);
516        Self::create_edge_ctx(&mut ctx, label, src, dst, props)
517    }
518
519    fn create_edge_ctx(
520        ctx: &mut WriteCtx,
521        label: &str,
522        src: NodeId,
523        dst: NodeId,
524        props: BTreeMap<String, PropertyValue>,
525    ) -> Result<EdgeId, GraphError> {
526        if ctx.nodes()?.get(src.0)?.is_none() {
527            return Err(GraphError::NodeNotFound(src));
528        }
529        if ctx.nodes()?.get(dst.0)?.is_none() {
530            return Err(GraphError::NodeNotFound(dst));
531        }
532        let label_id = intern_label(ctx, label)?;
533        let id = next_id(ctx, "next_edge_id")?;
534        let record = EdgeRecord {
535            label_id,
536            src: src.0,
537            dst: dst.0,
538            props,
539        };
540        let bytes = encode_edge(&record, |name| intern_prop(ctx, name))?;
541        ctx.edges()?.insert(id, bytes.as_slice())?;
542
543        ctx.adj_out()?
544            .insert(crate::model::adj_key(src.0, label_id, id), dst.0)?;
545        ctx.adj_in()?
546            .insert(crate::model::adj_key(dst.0, label_id, id), src.0)?;
547        Self::bump_rel_type_count(ctx, label_id, 1)?;
548        Ok(EdgeId(id))
549    }
550
551    /// Adjust `REL_TYPE_COUNTS` for one edge born (`+1`) or dying (`-1`)
552    /// -- called from the only two such places, `create_edge_ctx` and
553    /// `delete_edge_ctx`. Saturating on the way down: a file written by
554    /// a build that predates the table (or the backfill racing nothing
555    /// -- see `backfill_rel_type_counts`) must degrade to a wrong
556    /// *estimate*, never an underflow panic.
557    fn bump_rel_type_count(
558        ctx: &mut WriteCtx,
559        label_id: u32,
560        delta: i64,
561    ) -> Result<(), GraphError> {
562        let table = ctx.rel_type_counts()?;
563        let current = table.get(label_id)?.map(|g| g.value()).unwrap_or(0);
564        let next = current.saturating_add_signed(delta);
565        table.insert(label_id, next)?;
566        Ok(())
567    }
568
569    pub fn get_edge(&self, id: EdgeId) -> Result<Option<Edge>, GraphError> {
570        let read_txn = self.begin_read()?;
571        Self::get_edge_in_txn(Txn::Read(&read_txn), id)
572    }
573
574    pub fn get_edge_in_txn(txn: Txn, id: EdgeId) -> Result<Option<Edge>, GraphError> {
575        let record: Option<EdgeRecord> = {
576            let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
577            let found = match edges.get(id.0)? {
578                Some(guard) => {
579                    let mut resolve = prop_resolver(txn)?;
580                    Some(decode_edge(guard.value(), &mut resolve)?)
581                }
582                None => None,
583            };
584            found
585        };
586        let Some(record) = record else {
587            return Ok(None);
588        };
589        let label = resolve_label(txn, record.label_id)?;
590        Ok(Some(Edge {
591            id,
592            label,
593            src: NodeId(record.src),
594            dst: NodeId(record.dst),
595            props: record.props,
596        }))
597    }
598
599    /// Neighbors of `node` in `dir`, optionally filtered by edge label.
600    /// Reads directly from the adjacency multimap without touching `edges`.
601    pub fn neighbors(
602        &self,
603        node: NodeId,
604        dir: Direction,
605        label_filter: Option<&str>,
606    ) -> Result<Vec<AdjEntry>, GraphError> {
607        let read_txn = self.begin_read()?;
608        Self::neighbors_in_txn(Txn::Read(&read_txn), node, dir, label_filter)
609    }
610
611    pub fn neighbors_in_txn(
612        txn: Txn,
613        node: NodeId,
614        dir: Direction,
615        label_filter: Option<&str>,
616    ) -> Result<Vec<AdjEntry>, GraphError> {
617        // Typed expansion narrows the key range itself (`node ++ label`
618        // prefix) instead of post-filtering a full entry scan -- the
619        // O(matching degree) fix this composite key layout exists for.
620        let (lo, hi) = match label_filter {
621            Some(l) => match lookup_label_id(txn, l)? {
622                Some(lid) => crate::model::adj_label_bounds(node.0, lid),
623                None => return Ok(Vec::new()),
624            },
625            None => crate::model::adj_node_bounds(node.0),
626        };
627        let mut result = Vec::new();
628        let table_def = match dir {
629            Direction::Out => marsdb_storage::tables::ADJ_OUT,
630            Direction::In => marsdb_storage::tables::ADJ_IN,
631        };
632        let table = txn.open_table(table_def)?;
633        for item in table.range(lo..=hi)? {
634            let (key, value) = item?;
635            let (_, label_id, edge_id) = key.value();
636            result.push(AdjEntry {
637                edge_id: EdgeId(edge_id),
638                other: NodeId(value.value()),
639                label_id,
640            });
641        }
642        Ok(result)
643    }
644
645    pub fn delete_edge(&self, id: EdgeId) -> Result<bool, GraphError> {
646        let write_txn = self.begin_write()?;
647        let removed = Self::delete_edge_in_txn(&write_txn, id)?;
648        write_txn.commit()?;
649        Ok(removed)
650    }
651
652    pub fn delete_edge_in_txn(
653        write_txn: &WriteTransaction,
654        id: EdgeId,
655    ) -> Result<bool, GraphError> {
656        let mut ctx = WriteCtx::open(write_txn);
657        Ok(Self::delete_edge_ctx(&mut ctx, id)?.is_some())
658    }
659
660    /// Batch form of `delete_edge_in_txn`: one `WriteCtx` across every
661    /// id instead of a fresh one (and its table opens) per edge, and the
662    /// deleted edges' label names resolved here — once per distinct
663    /// label id, while the ctx is already open — instead of a separate
664    /// whole-edge fetch per id on the caller's side. Measured ~neutral
665    /// on wall time vs the per-edge path (a scattered bulk delete's cost
666    /// lives in the executor's match phase, not here — sorting the ids
667    /// into per-table passes was tried too and moved nothing), so this
668    /// exists for the API shape: one call for a `DELETE r` statement's
669    /// whole edge set, doing strictly less redundant work. Returns
670    /// `(id, label name)` for each edge that actually existed — an id
671    /// already gone (a duplicate in `ids`, or deleted by an earlier
672    /// statement) is silently skipped, same contract as the single-edge
673    /// form's `false`.
674    pub fn delete_edges_in_txn(
675        write_txn: &WriteTransaction,
676        ids: &[EdgeId],
677    ) -> Result<Vec<(EdgeId, String)>, GraphError> {
678        let mut ctx = WriteCtx::open(write_txn);
679        let mut label_names: HashMap<u32, String> = HashMap::new();
680        let mut deleted = Vec::with_capacity(ids.len());
681        for &id in ids {
682            let Some(label_id) = Self::delete_edge_ctx(&mut ctx, id)? else {
683                continue;
684            };
685            let name = match label_names.entry(label_id) {
686                std::collections::hash_map::Entry::Occupied(e) => e.get().clone(),
687                std::collections::hash_map::Entry::Vacant(e) => {
688                    let name = ctx
689                        .id_to_label()?
690                        .get(label_id)?
691                        .ok_or_else(|| {
692                            GraphError::CorruptData(format!(
693                                "label id {label_id} has no interned string"
694                            ))
695                        })?
696                        .value()
697                        .to_string();
698                    e.insert(name).clone()
699                }
700            };
701            deleted.push((id, name));
702        }
703        Ok(deleted)
704    }
705
706    /// Internal, `WriteCtx`-based logic -- `delete_node_in_txn` calls this
707    /// directly (not the public `delete_edge_in_txn` wrapper) for each of a
708    /// deleted node's incident edges, since it already has its own `ctx`
709    /// open for the same transaction; opening a second `WriteCtx` on top of
710    /// it would try to open every table twice and hit redb's
711    /// `TableAlreadyOpen`. Returns the deleted edge's label id, `None` if
712    /// the edge didn't exist.
713    fn delete_edge_ctx(ctx: &mut WriteCtx, id: EdgeId) -> Result<Option<u32>, GraphError> {
714        let Some(record_bytes) = ctx
715            .edges()?
716            .remove(id.0)?
717            .map(|guard| guard.value().to_vec())
718        else {
719            return Ok(None);
720        };
721        // Header-only read: adjacency cleanup needs (label, src, dst),
722        // never the edge's properties -- skips every prop-name resolution.
723        let (label_id, src, dst) = edge_header(&record_bytes)?;
724        ctx.adj_out()?
725            .remove(crate::model::adj_key(src, label_id, id.0))?;
726        ctx.adj_in()?
727            .remove(crate::model::adj_key(dst, label_id, id.0))?;
728        Self::bump_rel_type_count(ctx, label_id, -1)?;
729        Ok(Some(label_id))
730    }
731
732    /// Delete a node. If `detach` is false and the node has incident edges,
733    /// returns `GraphError::NodeHasEdges` instead of deleting anything.
734    pub fn delete_node(&self, id: NodeId, detach: bool) -> Result<bool, GraphError> {
735        let write_txn = self.begin_write()?;
736        let existed = Self::delete_node_in_txn(&write_txn, id, detach)?.is_some();
737        write_txn.commit()?;
738        Ok(existed)
739    }
740
741    /// Returns `None` when the node didn't exist, else
742    /// `Some(incident edges actually deleted)` — the caller-visible
743    /// count a `DETACH DELETE` needs for its statement stats (a
744    /// self-loop appears in both adjacency directions but deletes
745    /// once, so the count comes from the deletions, not the scan).
746    pub fn delete_node_in_txn(
747        write_txn: &WriteTransaction,
748        id: NodeId,
749        detach: bool,
750    ) -> Result<Option<u64>, GraphError> {
751        let mut ctx = WriteCtx::open(write_txn);
752        let mut incident: Vec<EdgeId> = Vec::new();
753        let (lo, hi) = crate::model::adj_node_bounds(id.0);
754        for item in ctx.adj_out()?.range(lo..=hi)? {
755            let (key, _) = item?;
756            let (_, _, edge_id) = key.value();
757            incident.push(EdgeId(edge_id));
758        }
759        for item in ctx.adj_in()?.range(lo..=hi)? {
760            let (key, _) = item?;
761            let (_, _, edge_id) = key.value();
762            incident.push(EdgeId(edge_id));
763        }
764        if !incident.is_empty() && !detach {
765            return Err(GraphError::NodeHasEdges(id));
766        }
767        let mut edges_deleted: u64 = 0;
768        for edge_id in incident {
769            if Self::delete_edge_ctx(&mut ctx, edge_id)?.is_some() {
770                edges_deleted += 1;
771            }
772        }
773        let Some(removed_bytes) = ctx
774            .nodes()?
775            .remove(id.0)?
776            .map(|guard| guard.value().to_vec())
777        else {
778            return Ok(None);
779        };
780        let record = decode_node(&removed_bytes, |pid| {
781            crate::index::resolve_prop_ctx(&mut ctx, pid)
782        })?;
783        for &label_id in &record.label_ids {
784            ctx.node_label_index()?.remove(label_id, id.0)?;
785        }
786        crate::index::on_node_deleted(&mut ctx, id.0, &record.label_ids, &record.props)?;
787        Ok(Some(edges_deleted))
788    }
789
790    /// Declares an index on `(label, prop)`, backfilling it from every
791    /// existing node with `label` — see `index::create_index`'s own docs
792    /// for the exact semantics (idempotency, unique-violation behavior).
793    pub fn create_index(&self, label: &str, prop: &str, unique: bool) -> Result<(), GraphError> {
794        let write_txn = self.begin_write()?;
795        Self::create_index_in_txn(&write_txn, label, prop, unique)?;
796        write_txn.commit()?;
797        Ok(())
798    }
799
800    /// Same as `create_index`, but against an already-open
801    /// `WriteTransaction` — for a caller (`CREATE INDEX` as a Cypher
802    /// statement) that's already inside one transaction and must commit
803    /// or abort it as a whole, not open a second one (redb allows only one
804    /// writer at a time; opening a second would deadlock).
805    pub fn create_index_in_txn(
806        write_txn: &WriteTransaction,
807        label: &str,
808        prop: &str,
809        unique: bool,
810    ) -> Result<(), GraphError> {
811        let mut ctx = WriteCtx::open(write_txn);
812        crate::index::create_index(&mut ctx, label, prop, unique)
813    }
814
815    /// `None` means no index is declared on `(label, prop)`.
816    pub fn index_def(
817        &self,
818        label: &str,
819        prop: &str,
820    ) -> Result<Option<crate::IndexDef>, GraphError> {
821        let read_txn = self.begin_read()?;
822        crate::index::lookup_index_def(Txn::Read(&read_txn), label, prop)
823    }
824
825    /// Same as `index_def`, but against an already-open `Txn` — for a
826    /// caller (the query planner/executor) that's already inside one
827    /// transaction and needs a consistent view, not a fresh snapshot.
828    pub fn index_def_in_txn(
829        txn: Txn,
830        label: &str,
831        prop: &str,
832    ) -> Result<Option<crate::IndexDef>, GraphError> {
833        crate::index::lookup_index_def(txn, label, prop)
834    }
835
836    /// Same as `lookup_by_index`, but against an already-open `Txn`.
837    pub fn lookup_by_index_in_txn(
838        txn: Txn,
839        label: &str,
840        prop: &str,
841        value: &PropertyValue,
842    ) -> Result<Vec<NodeId>, GraphError> {
843        crate::index::lookup_exact(txn, label, prop, value, None)
844    }
845
846    /// Same as `lookup_by_index_in_txn`, but stops once `limit` nodes are
847    /// found — the storage-level end of `LIMIT` push-down through an
848    /// `IndexSeek` (see `marsdb_query::planner`/`executor::stream_index_seek`).
849    /// Range counterpart of `lookup_by_index_in_txn` — every node whose
850    /// indexed value falls within the bounds (`(value, inclusive)` per
851    /// side, either side open). Returns a SUPERSET for numeric bounds
852    /// (int/float regions both scanned, lossy conversions widened
853    /// outward) — callers must re-check the original predicate; see
854    /// `index::lookup_range`.
855    pub fn lookup_by_index_range_in_txn(
856        txn: Txn,
857        label: &str,
858        prop: &str,
859        lo: Option<(&PropertyValue, bool)>,
860        hi: Option<(&PropertyValue, bool)>,
861        limit: Option<usize>,
862    ) -> Result<Vec<NodeId>, GraphError> {
863        crate::index::lookup_range(txn, label, prop, lo, hi, limit)
864    }
865
866    /// Resumable form of `lookup_by_index_range_in_txn` — see
867    /// `index::IndexRangeCursor` for the demand-driven contract.
868    pub fn index_range_cursor_in_txn(
869        txn: Txn,
870        label: &str,
871        prop: &str,
872        lo: Option<(&PropertyValue, bool)>,
873        hi: Option<(&PropertyValue, bool)>,
874    ) -> Result<Option<crate::index::IndexRangeCursor>, GraphError> {
875        crate::index::IndexRangeCursor::new(txn, label, prop, lo, hi)
876    }
877
878    pub fn lookup_by_index_limited_in_txn(
879        txn: Txn,
880        label: &str,
881        prop: &str,
882        value: &PropertyValue,
883        limit: usize,
884    ) -> Result<Vec<NodeId>, GraphError> {
885        crate::index::lookup_exact(txn, label, prop, value, Some(limit))
886    }
887
888    /// Cheap, exact count of nodes under `(label, prop) = value` — for the
889    /// query planner to compare selectivity between several indexed
890    /// equality candidates, not for fetching the nodes themselves (see
891    /// `lookup_by_index_in_txn`). O(1), same contract as `lookup_by_index`
892    /// re: "no index" vs "index, no match" both reading as `0`.
893    pub fn index_match_count_in_txn(
894        txn: Txn,
895        label: &str,
896        prop: &str,
897        value: &PropertyValue,
898    ) -> Result<u64, GraphError> {
899        crate::index::match_count(txn, label, prop, value)
900    }
901
902    /// Every node currently indexed under `(label, prop) = value`. Empty
903    /// (not an error) if no such index exists — check `index_def` first if
904    /// the caller needs to distinguish "no index" from "index, no match".
905    pub fn lookup_by_index(
906        &self,
907        label: &str,
908        prop: &str,
909        value: &PropertyValue,
910    ) -> Result<Vec<NodeId>, GraphError> {
911        let read_txn = self.begin_read()?;
912        crate::index::lookup_exact(Txn::Read(&read_txn), label, prop, value, None)
913    }
914
915    pub fn set_node_prop(
916        &self,
917        id: NodeId,
918        key: &str,
919        value: PropertyValue,
920    ) -> Result<bool, GraphError> {
921        let write_txn = self.begin_write()?;
922        let updated = Self::set_node_prop_in_txn(&write_txn, id, key, value)?;
923        if updated {
924            write_txn.commit()?;
925        } else {
926            write_txn.abort()?;
927        }
928        Ok(updated)
929    }
930
931    pub fn set_node_prop_in_txn(
932        write_txn: &WriteTransaction,
933        id: NodeId,
934        key: &str,
935        value: PropertyValue,
936    ) -> Result<bool, GraphError> {
937        let mut ctx = WriteCtx::open(write_txn);
938        let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
939            return Ok(false);
940        };
941        let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
942        let old_value = record.props.insert(key.to_string(), value.clone());
943        let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
944        ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
945        crate::index::on_node_prop_changed(
946            &mut ctx,
947            id.0,
948            &record.label_ids,
949            key,
950            old_value.as_ref(),
951            Some(&value),
952        )?;
953        Ok(true)
954    }
955
956    pub fn set_edge_prop(
957        &self,
958        id: EdgeId,
959        key: &str,
960        value: PropertyValue,
961    ) -> Result<bool, GraphError> {
962        let write_txn = self.begin_write()?;
963        let updated = Self::set_edge_prop_in_txn(&write_txn, id, key, value)?;
964        if updated {
965            write_txn.commit()?;
966        } else {
967            write_txn.abort()?;
968        }
969        Ok(updated)
970    }
971
972    pub fn set_edge_prop_in_txn(
973        write_txn: &WriteTransaction,
974        id: EdgeId,
975        key: &str,
976        value: PropertyValue,
977    ) -> Result<bool, GraphError> {
978        let mut ctx = WriteCtx::open(write_txn);
979        let Some(bytes) = ctx.edges()?.get(id.0)?.map(|g| g.value().to_vec()) else {
980            return Ok(false);
981        };
982        let mut record = decode_edge(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
983        record.props.insert(key.to_string(), value);
984        let new_bytes = encode_edge(&record, |name| intern_prop(&mut ctx, name))?;
985        ctx.edges()?.insert(id.0, new_bytes.as_slice())?;
986        Ok(true)
987    }
988
989    pub fn remove_node_prop_in_txn(
990        write_txn: &WriteTransaction,
991        id: NodeId,
992        key: &str,
993    ) -> Result<bool, GraphError> {
994        let mut ctx = WriteCtx::open(write_txn);
995        let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
996            return Ok(false);
997        };
998        let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
999        let old_value = record.props.remove(key);
1000        let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
1001        ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
1002        crate::index::on_node_prop_changed(
1003            &mut ctx,
1004            id.0,
1005            &record.label_ids,
1006            key,
1007            old_value.as_ref(),
1008            None,
1009        )?;
1010        Ok(true)
1011    }
1012
1013    pub fn remove_edge_prop_in_txn(
1014        write_txn: &WriteTransaction,
1015        id: EdgeId,
1016        key: &str,
1017    ) -> Result<bool, GraphError> {
1018        let mut ctx = WriteCtx::open(write_txn);
1019        let Some(bytes) = ctx.edges()?.get(id.0)?.map(|g| g.value().to_vec()) else {
1020            return Ok(false);
1021        };
1022        let mut record = decode_edge(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
1023        record.props.remove(key);
1024        let new_bytes = encode_edge(&record, |name| intern_prop(&mut ctx, name))?;
1025        ctx.edges()?.insert(id.0, new_bytes.as_slice())?;
1026        Ok(true)
1027    }
1028
1029    /// Adds `label` to `id`'s label set -- a no-op (not an error) if it's
1030    /// already there, same idempotent-add semantics real Cypher's `SET
1031    /// n:Label` has.
1032    pub fn add_node_label_in_txn(
1033        write_txn: &WriteTransaction,
1034        id: NodeId,
1035        label: &str,
1036    ) -> Result<bool, GraphError> {
1037        let mut ctx = WriteCtx::open(write_txn);
1038        let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
1039            return Ok(false);
1040        };
1041        let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
1042        let label_id = intern_label(&mut ctx, label)?;
1043        if !record.label_ids.contains(&label_id) {
1044            record.label_ids.push(label_id);
1045            let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
1046            ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
1047            ctx.node_label_index()?.insert(label_id, id.0)?;
1048            crate::index::on_node_created(&mut ctx, id.0, &[label_id], &record.props)?;
1049        }
1050        Ok(true)
1051    }
1052
1053    /// Removes `label` from `id`'s label set -- a no-op (not an error) if
1054    /// it's not there (label unknown entirely, or known but not on this
1055    /// node), same as real Cypher's `REMOVE n:Label`.
1056    pub fn remove_node_label_in_txn(
1057        write_txn: &WriteTransaction,
1058        id: NodeId,
1059        label: &str,
1060    ) -> Result<bool, GraphError> {
1061        let mut ctx = WriteCtx::open(write_txn);
1062        let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
1063            return Ok(false);
1064        };
1065        // Same lookup as `labels::lookup_label_id`, but reading directly
1066        // from the already-open `ctx.label_to_id` -- a second `Txn`-based
1067        // open of the same table would be `TableAlreadyOpen`.
1068        let Some(label_id) = ctx.label_to_id()?.get(label)?.map(|g| g.value()) else {
1069            return Ok(true);
1070        };
1071        let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
1072        if let Some(pos) = record.label_ids.iter().position(|&l| l == label_id) {
1073            record.label_ids.remove(pos);
1074            let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
1075            ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
1076            ctx.node_label_index()?.remove(label_id, id.0)?;
1077            crate::index::on_node_deleted(&mut ctx, id.0, &[label_id], &record.props)?;
1078        }
1079        Ok(true)
1080    }
1081
1082    /// Total node count — O(1) (redb tracks table entry counts). For the
1083    /// query planner's start-point cardinality comparison: the cost of an
1084    /// `AllNodesScan` leaf, never for fetching anything.
1085    pub fn node_count_in_txn(txn: Txn) -> Result<u64, GraphError> {
1086        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
1087        Ok(nodes.len()?)
1088    }
1089
1090    /// Number of nodes carrying `label` — O(1) via the label index's
1091    /// per-key entry count (same mechanism as `index_match_count_in_txn`).
1092    /// An unknown label reads as 0, same as everywhere else. Planner
1093    /// cardinality use only, like `node_count_in_txn`.
1094    pub fn label_count_in_txn(txn: Txn, label: &str) -> Result<u64, GraphError> {
1095        let Some(label_id) = lookup_label_id(txn, label)? else {
1096            return Ok(0);
1097        };
1098        let index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
1099        let count = index.get(label_id)?.len();
1100        Ok(count)
1101    }
1102
1103    /// Every interned name currently carried by at least one node, as
1104    /// `(label, node count)` sorted by label — the substance behind
1105    /// `CALL db.labels()`. Node labels and relationship types share one
1106    /// intern namespace (`intern_label` serves both), so membership here
1107    /// is decided by live *use* (a nonzero label-index count), not by
1108    /// interning: a name whose nodes were all deleted drops out, same as
1109    /// a name only ever used as a relationship type never appears.
1110    /// O(interned names), each with an O(1) count read.
1111    pub fn list_node_labels_in_txn(txn: Txn) -> Result<Vec<(String, u64)>, GraphError> {
1112        let l2i = txn.open_table(marsdb_storage::tables::LABEL_TO_ID)?;
1113        let index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
1114        let mut out = Vec::new();
1115        for entry in l2i.iter()? {
1116            let (name, id) = entry?;
1117            let count = index.get(id.value())?.len();
1118            if count > 0 {
1119                out.push((name.value().to_string(), count));
1120            }
1121        }
1122        Ok(out)
1123    }
1124
1125    /// Relationship-type counterpart of `list_node_labels_in_txn`:
1126    /// `(type, live edge count)` sorted by type, counts from
1127    /// `REL_TYPE_COUNTS`. Same live-use membership rule.
1128    pub fn list_rel_types_in_txn(txn: Txn) -> Result<Vec<(String, u64)>, GraphError> {
1129        let l2i = txn.open_table(marsdb_storage::tables::LABEL_TO_ID)?;
1130        let counts = txn.open_table(marsdb_storage::tables::REL_TYPE_COUNTS)?;
1131        let mut out = Vec::new();
1132        for entry in l2i.iter()? {
1133            let (name, id) = entry?;
1134            let count = counts.get(id.value())?.map(|g| g.value()).unwrap_or(0);
1135            if count > 0 {
1136                out.push((name.value().to_string(), count));
1137            }
1138        }
1139        Ok(out)
1140    }
1141
1142    /// Every interned property name, sorted — `CALL db.propertyKeys()`.
1143    /// Interning is permanent (there is no un-intern on last use, unlike
1144    /// the liveness rule above, which has cheap per-name counts to
1145    /// consult), so this lists every key that has ever appeared.
1146    pub fn list_property_keys_in_txn(txn: Txn) -> Result<Vec<String>, GraphError> {
1147        let p2i = txn.open_table(marsdb_storage::tables::PROP_TO_ID)?;
1148        let mut out = Vec::new();
1149        for entry in p2i.iter()? {
1150            let (name, _) = entry?;
1151            out.push(name.value().to_string());
1152        }
1153        Ok(out)
1154    }
1155
1156    /// Every declared index as `(label, property, unique)` —
1157    /// `CALL db.indexes()`. Full `INDEX_DEFS` scan; the number of
1158    /// declared indexes is small by nature.
1159    pub fn list_indexes_in_txn(txn: Txn) -> Result<Vec<(String, String, bool)>, GraphError> {
1160        let mut resolve_prop = prop_resolver(txn)?;
1161        let defs = txn.open_table(marsdb_storage::tables::INDEX_DEFS)?;
1162        let mut out = Vec::new();
1163        for entry in defs.iter()? {
1164            let (key, value) = entry?;
1165            let key_bytes = key.value();
1166            let label_id = u32::from_be_bytes(key_bytes[0..4].try_into().map_err(|_| {
1167                GraphError::CorruptData("index key prefix shorter than 8 bytes".into())
1168            })?);
1169            let prop_id = u32::from_be_bytes(key_bytes[4..8].try_into().map_err(|_| {
1170                GraphError::CorruptData("index key prefix shorter than 8 bytes".into())
1171            })?);
1172            let def: crate::IndexDef = postcard::from_bytes(value.value())
1173                .map_err(|e| GraphError::CorruptData(format!("undecodable index def: {e}")))?;
1174            out.push((
1175                resolve_label(txn, label_id)?,
1176                resolve_prop(prop_id)?,
1177                def.unique,
1178            ));
1179        }
1180        Ok(out)
1181    }
1182
1183    /// Resolve a label/relationship-type name to its interned id —
1184    /// `None` if never interned. Scan-support API (`EdgeScanCursor`
1185    /// consumers pre-resolve type names once per scan).
1186    pub fn label_id_for(txn: Txn, name: &str) -> Result<Option<u32>, GraphError> {
1187        lookup_label_id(txn, name)
1188    }
1189
1190    /// Header fields `(label_id, src, dst)` of a raw edge record as
1191    /// returned by `EdgeScanCursor` — no property decode.
1192    pub fn edge_record_header(bytes: &[u8]) -> Result<(u32, u64, u64), GraphError> {
1193        edge_header(bytes)
1194    }
1195
1196    /// One property's value from a raw edge record, by interned prop
1197    /// id — a directory-entry read from in-hand bytes, no storage
1198    /// access. `Ok(None)` = property absent on this edge.
1199    pub fn edge_record_prop(
1200        bytes: &[u8],
1201        prop_id: u32,
1202    ) -> Result<Option<PropertyValue>, GraphError> {
1203        match crate::encode::edge_prop_raw(bytes, prop_id)? {
1204            Some(raw) => Ok(Some(crate::encode::decode_value(raw)?)),
1205            None => Ok(None),
1206        }
1207    }
1208
1209    /// Resumable chunked sweep over the whole `EDGES` table in id
1210    /// order — the sequential-scan primitive behind the planner's
1211    /// `EdgeTypeScan`. Same demand-driven shape as `IndexRangeCursor`:
1212    /// each `next_chunk` re-seeks past the last returned id (O(log n))
1213    /// and copies at most `chunk_size` raw records out, so a `LIMIT`ed
1214    /// consumer that stops early never pays for the rest of the table.
1215    pub fn edge_scan_cursor() -> EdgeScanCursor {
1216        EdgeScanCursor { resume_after: None }
1217    }
1218
1219    /// Total edge count — O(1) (redb tracks table entry counts), the
1220    /// edge counterpart of `node_count_in_txn`. Planner cardinality use
1221    /// only.
1222    pub fn edge_count_in_txn(txn: Txn) -> Result<u64, GraphError> {
1223        let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
1224        Ok(edges.len()?)
1225    }
1226
1227    /// Number of live edges of relationship type `rel_type` — O(1) via
1228    /// `REL_TYPE_COUNTS` (see its definition in `tables.rs` for the
1229    /// maintenance/backfill story). An unknown type reads as 0, same as
1230    /// `label_count_in_txn`. Planner cardinality use only.
1231    pub fn rel_type_count_in_txn(txn: Txn, rel_type: &str) -> Result<u64, GraphError> {
1232        let Some(label_id) = lookup_label_id(txn, rel_type)? else {
1233            return Ok(0);
1234        };
1235        let counts = txn.open_table(marsdb_storage::tables::REL_TYPE_COUNTS)?;
1236        let count = counts.get(label_id)?.map(|g| g.value()).unwrap_or(0);
1237        Ok(count)
1238    }
1239
1240    /// Full scan of all nodes, optionally filtered by label. v1 has no
1241    /// secondary index on label, so this is a linear scan of the table.
1242    pub fn all_nodes(&self, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
1243        let read_txn = self.begin_read()?;
1244        Self::all_nodes_in_txn(Txn::Read(&read_txn), label_filter)
1245    }
1246
1247    /// Scan only graph identities, without decoding node records. Query
1248    /// pipelines use this to defer record/property loading until a filter or
1249    /// projection actually needs it.
1250    pub fn all_node_ids_limited_in_txn(
1251        txn: Txn,
1252        label_filter: Option<&str>,
1253        limit: usize,
1254    ) -> Result<Vec<NodeId>, GraphError> {
1255        let Some(label_filter) = label_filter else {
1256            let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
1257            return nodes
1258                .iter()?
1259                .take(limit)
1260                .map(|entry| {
1261                    entry
1262                        .map(|(key, _)| NodeId(key.value()))
1263                        .map_err(Into::into)
1264                })
1265                .collect();
1266        };
1267        let Some(label_id) = lookup_label_id(txn, label_filter)? else {
1268            return Ok(Vec::new());
1269        };
1270        let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
1271        let ids = label_index
1272            .get(label_id)?
1273            .take(limit)
1274            .map(|entry| entry.map(|value| NodeId(value.value())).map_err(Into::into))
1275            .collect::<Result<Vec<_>, GraphError>>()?;
1276        drop(label_index);
1277        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
1278        for id in &ids {
1279            if nodes.get(id.0)?.is_none() {
1280                return Err(GraphError::CorruptData(format!(
1281                    "node label index references missing node {}",
1282                    id.0
1283                )));
1284            }
1285        }
1286        Ok(ids)
1287    }
1288
1289    pub fn all_nodes_in_txn(txn: Txn, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
1290        Self::all_nodes_limited_in_txn(txn, label_filter, usize::MAX)
1291    }
1292
1293    /// Same as `all_nodes_in_txn`, but stops once `limit` nodes are found --
1294    /// the storage-level end of `LIMIT` push-down (see the executor's
1295    /// `scan()`/`eval_plan` docs for the query-level half): a query whose
1296    /// entire plan is a bare scan feeding straight into a `LIMIT` doesn't
1297    /// need to touch rows past the first `limit`, whether or not a label
1298    /// filter narrows it first.
1299    pub fn all_nodes_limited_in_txn(
1300        txn: Txn,
1301        label_filter: Option<&str>,
1302        limit: usize,
1303    ) -> Result<Vec<Node>, GraphError> {
1304        // A label filter goes through NODE_LABEL_INDEX (label_id -> node_ids)
1305        // plus a point lookup per match, instead of scanning every row in
1306        // NODES — cost scales with the number of matching rows, not the
1307        // table size. No filter means every row is wanted anyway, so a full
1308        // scan is already optimal; the index wouldn't help.
1309        let Some(label_filter) = label_filter else {
1310            let mut result = Vec::new();
1311            let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
1312            // Resolver hoisted out of the loop: one ID_TO_PROP open for the
1313            // whole scan, not one per record (table opens were themselves a
1314            // measured hot cost -- mars-3va).
1315            let mut resolve = prop_resolver(txn)?;
1316            for item in nodes.iter()? {
1317                if result.len() >= limit {
1318                    break;
1319                }
1320                let (key, value) = item?;
1321                let record = decode_node(value.value(), &mut resolve)?;
1322                let labels = record
1323                    .label_ids
1324                    .iter()
1325                    .map(|&lid| resolve_label(txn, lid))
1326                    .collect::<Result<Vec<_>, _>>()?;
1327                result.push(Node {
1328                    id: NodeId(key.value()),
1329                    labels,
1330                    props: record.props,
1331                });
1332            }
1333            return Ok(result);
1334        };
1335        let Some(label_id) = lookup_label_id(txn, label_filter)? else {
1336            return Ok(Vec::new());
1337        };
1338        let node_ids: Vec<u64> = {
1339            let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
1340            // `.take(limit)` here, not a `.truncate()` after collecting --
1341            // stops walking the multimap's own entries past `limit`, not
1342            // just the (more expensive) per-id NODES point-reads below.
1343            // Measured difference: without this, a labeled LIMIT query's
1344            // cost still scaled with the *matching* row count, not `limit`
1345            // (see BENCHMARKS.md's `execute_scan_limit_pushdown` numbers).
1346            let ids: Vec<u64> = label_index
1347                .get(label_id)?
1348                .take(limit)
1349                .map(|item| item.map(|g| g.value()))
1350                .collect::<Result<_, _>>()?;
1351            ids
1352        };
1353        let mut result = Vec::with_capacity(node_ids.len());
1354        let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
1355        // Same loop-hoisted resolver as the unfiltered scan above.
1356        let mut resolve = prop_resolver(txn)?;
1357        for id in node_ids {
1358            let guard = nodes.get(id)?.ok_or_else(|| {
1359                GraphError::CorruptData(format!("node label index references missing node {}", id))
1360            })?;
1361            let record = decode_node(guard.value(), &mut resolve)?;
1362            drop(guard);
1363            let labels = record
1364                .label_ids
1365                .iter()
1366                .map(|&lid| resolve_label(txn, lid))
1367                .collect::<Result<Vec<_>, _>>()?;
1368            result.push(Node {
1369                id: NodeId(id),
1370                labels,
1371                props: record.props,
1372            });
1373        }
1374        Ok(result)
1375    }
1376}
1377
1378/// See `GraphStore::edge_scan_cursor`.
1379pub struct EdgeScanCursor {
1380    resume_after: Option<u64>,
1381}
1382
1383impl EdgeScanCursor {
1384    /// At most `chunk_size` `(edge_id, raw record bytes)` pairs, id
1385    /// order, starting after the previous chunk's last id. Empty vec =
1386    /// table exhausted.
1387    pub fn next_chunk(
1388        &mut self,
1389        txn: Txn,
1390        chunk_size: usize,
1391    ) -> Result<Vec<(u64, Vec<u8>)>, GraphError> {
1392        if chunk_size == 0 {
1393            return Ok(Vec::new());
1394        }
1395        let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
1396        let mut out = Vec::with_capacity(chunk_size.min(1024));
1397        let iter = match self.resume_after {
1398            Some(last) => {
1399                edges.range::<u64>((std::ops::Bound::Excluded(last), std::ops::Bound::Unbounded))?
1400            }
1401            None => edges.iter()?,
1402        };
1403        for entry in iter {
1404            let (id, value) = entry?;
1405            let id = id.value();
1406            out.push((id, value.value().to_vec()));
1407            self.resume_after = Some(id);
1408            if out.len() >= chunk_size {
1409                break;
1410            }
1411        }
1412        Ok(out)
1413    }
1414}
1415
1416#[cfg(test)]
1417mod tests {
1418    use super::*;
1419
1420    #[test]
1421    fn integrity_check_rejects_missing_node_label_index_entry() {
1422        let mut store = GraphStore::open_memory().unwrap();
1423        let node = store.create_node(&["Person"], BTreeMap::new()).unwrap();
1424
1425        let write = store.begin_write().unwrap();
1426        let label_id = {
1427            let labels = write
1428                .open_table(marsdb_storage::tables::LABEL_TO_ID)
1429                .unwrap();
1430            let id = labels.get("Person").unwrap().unwrap().value();
1431            id
1432        };
1433        write
1434            .open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)
1435            .unwrap()
1436            .remove(label_id, node.0)
1437            .unwrap();
1438        write.commit().unwrap();
1439
1440        let error = store.check_integrity().unwrap_err();
1441        assert!(
1442            matches!(error, GraphError::CorruptData(message) if message.contains("missing from the label index"))
1443        );
1444    }
1445
1446    #[test]
1447    fn integrity_check_rejects_dangling_adjacency_entry() {
1448        let mut store = GraphStore::open_memory().unwrap();
1449        let node = store.create_node(&[], BTreeMap::new()).unwrap();
1450
1451        let write = store.begin_write().unwrap();
1452        write
1453            .open_table(marsdb_storage::tables::ADJ_OUT)
1454            .unwrap()
1455            .insert(crate::model::adj_key(node.0, 0, 999), node.0)
1456            .unwrap();
1457        write.commit().unwrap();
1458
1459        let error = store.check_integrity().unwrap_err();
1460        assert!(
1461            matches!(error, GraphError::CorruptData(message) if message.contains("missing edge 999"))
1462        );
1463    }
1464
1465    #[test]
1466    fn create_index_backfills_existing_nodes() {
1467        let store = GraphStore::open_memory().unwrap();
1468        let mut alice_props = BTreeMap::new();
1469        alice_props.insert(
1470            "email".to_string(),
1471            PropertyValue::String("alice@x.com".to_string()),
1472        );
1473        let alice = store.create_node(&["Person"], alice_props).unwrap();
1474        let mut bob_props = BTreeMap::new();
1475        bob_props.insert(
1476            "email".to_string(),
1477            PropertyValue::String("bob@x.com".to_string()),
1478        );
1479        store.create_node(&["Person"], bob_props).unwrap();
1480        // A Person with no email at all -- must not show up under any lookup.
1481        store.create_node(&["Person"], BTreeMap::new()).unwrap();
1482
1483        store.create_index("Person", "email", false).unwrap();
1484
1485        let found = store
1486            .lookup_by_index(
1487                "Person",
1488                "email",
1489                &PropertyValue::String("alice@x.com".to_string()),
1490            )
1491            .unwrap();
1492        assert_eq!(found, vec![alice]);
1493    }
1494
1495    #[test]
1496    fn create_index_rejects_duplicate_unique_value() {
1497        let store = GraphStore::open_memory().unwrap();
1498        let mut props1 = BTreeMap::new();
1499        props1.insert(
1500            "email".to_string(),
1501            PropertyValue::String("same@x.com".to_string()),
1502        );
1503        store.create_node(&["Person"], props1).unwrap();
1504        let mut props2 = BTreeMap::new();
1505        props2.insert(
1506            "email".to_string(),
1507            PropertyValue::String("same@x.com".to_string()),
1508        );
1509        store.create_node(&["Person"], props2).unwrap();
1510
1511        let error = store.create_index("Person", "email", true).unwrap_err();
1512        assert!(matches!(
1513            error,
1514            GraphError::UniqueConstraintViolation { .. }
1515        ));
1516
1517        // A rejected unique index must not partially exist.
1518        assert!(store.index_def("Person", "email").unwrap().is_none());
1519    }
1520
1521    #[test]
1522    fn lookup_by_index_on_undeclared_index_is_empty_not_an_error() {
1523        let store = GraphStore::open_memory().unwrap();
1524        store.create_node(&["Person"], BTreeMap::new()).unwrap();
1525        let found = store
1526            .lookup_by_index("Person", "email", &PropertyValue::String("x".to_string()))
1527            .unwrap();
1528        assert_eq!(found, Vec::new());
1529        assert!(store.index_def("Person", "email").unwrap().is_none());
1530    }
1531
1532    #[test]
1533    fn index_survives_reopen() {
1534        let dir = tempfile::tempdir().unwrap();
1535        let path = dir.path().join("index.db");
1536        {
1537            let store = GraphStore::open_file(&path).unwrap();
1538            let mut props = BTreeMap::new();
1539            props.insert(
1540                "email".to_string(),
1541                PropertyValue::String("x@x.com".to_string()),
1542            );
1543            store.create_node(&["Person"], props).unwrap();
1544            store.create_index("Person", "email", false).unwrap();
1545        }
1546        let store = GraphStore::open_file(&path).unwrap();
1547        assert!(store.index_def("Person", "email").unwrap().is_some());
1548        let found = store
1549            .lookup_by_index(
1550                "Person",
1551                "email",
1552                &PropertyValue::String("x@x.com".to_string()),
1553            )
1554            .unwrap();
1555        assert_eq!(found.len(), 1);
1556    }
1557
1558    #[test]
1559    fn create_node_after_index_declared_is_indexed_immediately() {
1560        let store = GraphStore::open_memory().unwrap();
1561        store.create_index("Person", "email", false).unwrap();
1562        let mut props = BTreeMap::new();
1563        props.insert(
1564            "email".to_string(),
1565            PropertyValue::String("new@x.com".to_string()),
1566        );
1567        let node = store.create_node(&["Person"], props).unwrap();
1568
1569        let found = store
1570            .lookup_by_index(
1571                "Person",
1572                "email",
1573                &PropertyValue::String("new@x.com".to_string()),
1574            )
1575            .unwrap();
1576        assert_eq!(found, vec![node]);
1577    }
1578
1579    #[test]
1580    fn set_node_prop_moves_the_index_entry() {
1581        let store = GraphStore::open_memory().unwrap();
1582        let mut props = BTreeMap::new();
1583        props.insert(
1584            "email".to_string(),
1585            PropertyValue::String("old@x.com".to_string()),
1586        );
1587        let node = store.create_node(&["Person"], props).unwrap();
1588        store.create_index("Person", "email", false).unwrap();
1589
1590        store
1591            .set_node_prop(
1592                node,
1593                "email",
1594                PropertyValue::String("new@x.com".to_string()),
1595            )
1596            .unwrap();
1597
1598        assert!(store
1599            .lookup_by_index(
1600                "Person",
1601                "email",
1602                &PropertyValue::String("old@x.com".to_string())
1603            )
1604            .unwrap()
1605            .is_empty());
1606        assert_eq!(
1607            store
1608                .lookup_by_index(
1609                    "Person",
1610                    "email",
1611                    &PropertyValue::String("new@x.com".to_string())
1612                )
1613                .unwrap(),
1614            vec![node]
1615        );
1616    }
1617
1618    #[test]
1619    fn set_node_prop_enforces_unique_index() {
1620        let store = GraphStore::open_memory().unwrap();
1621        let mut props1 = BTreeMap::new();
1622        props1.insert(
1623            "email".to_string(),
1624            PropertyValue::String("a@x.com".to_string()),
1625        );
1626        store.create_node(&["Person"], props1).unwrap();
1627        let mut props2 = BTreeMap::new();
1628        props2.insert(
1629            "email".to_string(),
1630            PropertyValue::String("b@x.com".to_string()),
1631        );
1632        let node2 = store.create_node(&["Person"], props2).unwrap();
1633        store.create_index("Person", "email", true).unwrap();
1634
1635        let error = store
1636            .set_node_prop(node2, "email", PropertyValue::String("a@x.com".to_string()))
1637            .unwrap_err();
1638        assert!(matches!(
1639            error,
1640            GraphError::UniqueConstraintViolation { .. }
1641        ));
1642    }
1643
1644    #[test]
1645    fn remove_node_prop_removes_the_index_entry() {
1646        let store = GraphStore::open_memory().unwrap();
1647        let mut props = BTreeMap::new();
1648        props.insert(
1649            "email".to_string(),
1650            PropertyValue::String("gone@x.com".to_string()),
1651        );
1652        let node = store.create_node(&["Person"], props).unwrap();
1653        store.create_index("Person", "email", false).unwrap();
1654
1655        let write = store.begin_write().unwrap();
1656        GraphStore::remove_node_prop_in_txn(&write, node, "email").unwrap();
1657        write.commit().unwrap();
1658
1659        assert!(store
1660            .lookup_by_index(
1661                "Person",
1662                "email",
1663                &PropertyValue::String("gone@x.com".to_string())
1664            )
1665            .unwrap()
1666            .is_empty());
1667    }
1668
1669    #[test]
1670    fn delete_node_removes_its_index_entries() {
1671        let store = GraphStore::open_memory().unwrap();
1672        let mut props = BTreeMap::new();
1673        props.insert(
1674            "email".to_string(),
1675            PropertyValue::String("deleted@x.com".to_string()),
1676        );
1677        let node = store.create_node(&["Person"], props).unwrap();
1678        store.create_index("Person", "email", false).unwrap();
1679
1680        store.delete_node(node, false).unwrap();
1681
1682        assert!(store
1683            .lookup_by_index(
1684                "Person",
1685                "email",
1686                &PropertyValue::String("deleted@x.com".to_string())
1687            )
1688            .unwrap()
1689            .is_empty());
1690    }
1691
1692    #[test]
1693    fn add_node_label_indexes_existing_props_under_the_new_label() {
1694        let store = GraphStore::open_memory().unwrap();
1695        let mut props = BTreeMap::new();
1696        props.insert(
1697            "email".to_string(),
1698            PropertyValue::String("multi@x.com".to_string()),
1699        );
1700        let node = store.create_node(&["Contact"], props).unwrap();
1701        store.create_index("Person", "email", false).unwrap();
1702
1703        // Not indexed yet -- the node isn't a Person.
1704        assert!(store
1705            .lookup_by_index(
1706                "Person",
1707                "email",
1708                &PropertyValue::String("multi@x.com".to_string())
1709            )
1710            .unwrap()
1711            .is_empty());
1712
1713        let write = store.begin_write().unwrap();
1714        GraphStore::add_node_label_in_txn(&write, node, "Person").unwrap();
1715        write.commit().unwrap();
1716
1717        assert_eq!(
1718            store
1719                .lookup_by_index(
1720                    "Person",
1721                    "email",
1722                    &PropertyValue::String("multi@x.com".to_string())
1723                )
1724                .unwrap(),
1725            vec![node]
1726        );
1727    }
1728
1729    #[test]
1730    fn remove_node_label_removes_index_entries_under_that_label() {
1731        let store = GraphStore::open_memory().unwrap();
1732        let mut props = BTreeMap::new();
1733        props.insert(
1734            "email".to_string(),
1735            PropertyValue::String("dual@x.com".to_string()),
1736        );
1737        let node = store.create_node(&["Person", "Contact"], props).unwrap();
1738        store.create_index("Person", "email", false).unwrap();
1739        assert_eq!(
1740            store
1741                .lookup_by_index(
1742                    "Person",
1743                    "email",
1744                    &PropertyValue::String("dual@x.com".to_string())
1745                )
1746                .unwrap(),
1747            vec![node]
1748        );
1749
1750        let write = store.begin_write().unwrap();
1751        GraphStore::remove_node_label_in_txn(&write, node, "Person").unwrap();
1752        write.commit().unwrap();
1753
1754        assert!(store
1755            .lookup_by_index(
1756                "Person",
1757                "email",
1758                &PropertyValue::String("dual@x.com".to_string())
1759            )
1760            .unwrap()
1761            .is_empty());
1762    }
1763}