use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use marsdb_storage::{
ReadTransaction, ReadableMultimapTable, ReadableTable, StorageEngine, Txn, WriteTransaction,
};
use crate::encode::{
decode_edge, decode_node, edge_header, encode_edge, encode_node, node_label_ids, EdgeRecord,
NodeRecord,
};
use crate::error::GraphError;
use crate::id::next_id;
use crate::labels::{intern_label, lookup_label_id, resolve_label};
use crate::model::{AdjEntry, Direction, Edge, EdgeId, Node, NodeId, PropertyValue};
use crate::props::{intern_prop, prop_resolver};
use crate::write_ctx::WriteCtx;
pub struct GraphStore {
storage: StorageEngine,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IntegrityReport {
pub physical_was_clean: bool,
pub labels: u64,
pub nodes: u64,
pub edges: u64,
}
impl GraphStore {
pub fn open_file(path: impl AsRef<Path>) -> Result<Self, GraphError> {
Ok(Self {
storage: StorageEngine::open_file(path)?,
})
}
pub fn open_memory() -> Result<Self, GraphError> {
Ok(Self {
storage: StorageEngine::open_memory()?,
})
}
pub fn backup_to(&self, path: impl AsRef<Path>) -> Result<(), GraphError> {
self.storage.backup_to(path)?;
Ok(())
}
pub fn check_integrity(&mut self) -> Result<IntegrityReport, GraphError> {
let physical_was_clean = self.storage.check_integrity()?;
let read = self.storage.begin_read()?;
let mut labels_by_id = BTreeMap::new();
{
let table = read.open_table(marsdb_storage::tables::ID_TO_LABEL)?;
for entry in table.iter()? {
let (id, label) = entry?;
labels_by_id.insert(id.value(), label.value().to_owned());
}
}
{
let table = read.open_table(marsdb_storage::tables::LABEL_TO_ID)?;
let mut count = 0usize;
for entry in table.iter()? {
let (label, id) = entry?;
count += 1;
if labels_by_id.get(&id.value()).map(String::as_str) != Some(label.value()) {
return Err(GraphError::CorruptData(format!(
"label mapping {:?} -> {} has no matching reverse mapping",
label.value(),
id.value()
)));
}
}
if count != labels_by_id.len() {
return Err(GraphError::CorruptData(
"label mapping tables have different entry counts".into(),
));
}
}
let mut nodes = BTreeMap::<u64, Vec<u32>>::new();
{
let table = read.open_table(marsdb_storage::tables::NODES)?;
for entry in table.iter()? {
let (id, value) = entry?;
let label_ids = node_label_ids(value.value())?;
for label_id in &label_ids {
if !labels_by_id.contains_key(label_id) {
return Err(GraphError::CorruptData(format!(
"node {} references unknown label {}",
id.value(),
label_id
)));
}
}
nodes.insert(id.value(), label_ids);
}
}
let mut indexed_labels = BTreeSet::new();
{
let table = read.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
for entry in table.iter()? {
let (label_id, values) = entry?;
let label_id = label_id.value();
if !labels_by_id.contains_key(&label_id) {
return Err(GraphError::CorruptData(format!(
"node label index references unknown label {label_id}"
)));
}
for node_id in values {
let node_id = node_id?.value();
let Some(node_labels) = nodes.get(&node_id) else {
return Err(GraphError::CorruptData(format!(
"node label index references missing node {node_id}"
)));
};
if !node_labels.contains(&label_id) {
return Err(GraphError::CorruptData(format!(
"node label index has label {label_id} for node {node_id}, but the node does not"
)));
}
indexed_labels.insert((label_id, node_id));
}
}
}
for (node_id, label_ids) in &nodes {
for label_id in label_ids {
if !indexed_labels.contains(&(*label_id, *node_id)) {
return Err(GraphError::CorruptData(format!(
"node {node_id} has label {label_id} but is missing from the label index"
)));
}
}
}
let mut edges = BTreeMap::<u64, (u32, u64, u64)>::new();
{
let table = read.open_table(marsdb_storage::tables::EDGES)?;
for entry in table.iter()? {
let (id, value) = entry?;
let (label_id, src, dst) = edge_header(value.value())?;
if !labels_by_id.contains_key(&label_id) {
return Err(GraphError::CorruptData(format!(
"edge {} references unknown label {}",
id.value(),
label_id
)));
}
if !nodes.contains_key(&src) || !nodes.contains_key(&dst) {
return Err(GraphError::CorruptData(format!(
"edge {} references missing endpoint {} -> {}",
id.value(),
src,
dst
)));
}
edges.insert(id.value(), (label_id, src, dst));
}
}
let outgoing =
Self::check_adjacency(&read, marsdb_storage::tables::ADJ_OUT, &nodes, &edges, true)?;
let incoming =
Self::check_adjacency(&read, marsdb_storage::tables::ADJ_IN, &nodes, &edges, false)?;
for (&edge_id, &(label_id, src, dst)) in &edges {
if !outgoing.contains(&(src, edge_id, dst, label_id)) {
return Err(GraphError::CorruptData(format!(
"edge {edge_id} is missing from outgoing adjacency"
)));
}
if !incoming.contains(&(dst, edge_id, src, label_id)) {
return Err(GraphError::CorruptData(format!(
"edge {edge_id} is missing from incoming adjacency"
)));
}
}
let meta = read.open_table(marsdb_storage::tables::META)?;
for (counter, maximum) in [
("next_node_id", nodes.keys().next_back().copied()),
("next_edge_id", edges.keys().next_back().copied()),
] {
if let Some(maximum) = maximum {
let stored = meta.get(counter)?.map(|value| value.value()).unwrap_or(0);
if stored < maximum {
return Err(GraphError::CorruptData(format!(
"{counter} counter {stored} is below maximum allocated id {maximum}"
)));
}
}
}
Ok(IntegrityReport {
physical_was_clean,
labels: labels_by_id.len() as u64,
nodes: nodes.len() as u64,
edges: edges.len() as u64,
})
}
fn check_adjacency(
read: &ReadTransaction,
definition: marsdb_storage::TableDefinition<(u64, u32, u64), u64>,
nodes: &BTreeMap<u64, Vec<u32>>,
edges: &BTreeMap<u64, (u32, u64, u64)>,
outgoing: bool,
) -> Result<BTreeSet<(u64, u64, u64, u32)>, GraphError> {
let table = read.open_table(definition)?;
let mut found = BTreeSet::new();
for entry in table.iter()? {
let (key, value) = entry?;
let (owner, key_label_id, edge_id) = key.value();
let other = value.value();
if !nodes.contains_key(&owner) {
return Err(GraphError::CorruptData(format!(
"adjacency references missing owner node {owner}"
)));
}
let Some(&(label_id, src, dst)) = edges.get(&edge_id) else {
return Err(GraphError::CorruptData(format!(
"adjacency references missing edge {edge_id}"
)));
};
let expected = if outgoing { (src, dst) } else { (dst, src) };
if owner != expected.0 || other != expected.1 || key_label_id != label_id {
return Err(GraphError::CorruptData(format!(
"adjacency entry for edge {edge_id} does not match the edge record"
)));
}
found.insert((owner, edge_id, other, key_label_id));
}
Ok(found)
}
pub fn begin_write(&self) -> Result<WriteTransaction, GraphError> {
Ok(self.storage.begin_write()?)
}
pub fn begin_read(&self) -> Result<ReadTransaction, GraphError> {
Ok(self.storage.begin_read()?)
}
pub fn commit(write_txn: WriteTransaction) -> Result<(), GraphError> {
write_txn.commit()?;
Ok(())
}
pub fn abort(write_txn: WriteTransaction) -> Result<(), GraphError> {
write_txn.abort()?;
Ok(())
}
pub fn create_node(
&self,
labels: &[&str],
props: BTreeMap<String, PropertyValue>,
) -> Result<NodeId, GraphError> {
let write_txn = self.begin_write()?;
let id = Self::create_node_in_txn(&write_txn, labels, props)?;
write_txn.commit()?;
Ok(id)
}
pub fn create_node_in_txn(
write_txn: &WriteTransaction,
labels: &[&str],
props: BTreeMap<String, PropertyValue>,
) -> Result<NodeId, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
Self::create_node_ctx(&mut ctx, labels, props)
}
fn create_node_ctx(
ctx: &mut WriteCtx,
labels: &[&str],
props: BTreeMap<String, PropertyValue>,
) -> Result<NodeId, GraphError> {
let label_ids = labels
.iter()
.map(|l| intern_label(ctx, l))
.collect::<Result<Vec<_>, _>>()?;
let id = next_id(ctx, "next_node_id")?;
let record = NodeRecord {
label_ids: label_ids.clone(),
props,
};
let bytes = encode_node(&record, |name| intern_prop(ctx, name))?;
ctx.nodes()?.insert(id, bytes.as_slice())?;
for &label_id in &label_ids {
ctx.node_label_index()?.insert(label_id, id)?;
}
crate::index::on_node_created(ctx, id, &label_ids, &record.props)?;
Ok(NodeId(id))
}
pub fn get_node(&self, id: NodeId) -> Result<Option<Node>, GraphError> {
let read_txn = self.begin_read()?;
Self::get_node_in_txn(Txn::Read(&read_txn), id)
}
pub fn get_node_in_txn(txn: Txn, id: NodeId) -> Result<Option<Node>, GraphError> {
let record: Option<NodeRecord> = {
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
let found = match nodes.get(id.0)? {
Some(guard) => {
let mut resolve = prop_resolver(txn)?;
Some(decode_node(guard.value(), &mut resolve)?)
}
None => None,
};
found
};
let Some(record) = record else {
return Ok(None);
};
let labels = record
.label_ids
.iter()
.map(|&lid| resolve_label(txn, lid))
.collect::<Result<Vec<_>, _>>()?;
Ok(Some(Node {
id,
labels,
props: record.props,
}))
}
pub fn lookup_prop_id_in_txn(txn: Txn, prop: &str) -> Result<Option<u32>, GraphError> {
crate::props::lookup_prop_id(txn, prop)
}
pub fn get_node_prop_in_txn(
txn: Txn,
id: NodeId,
prop_id: u32,
) -> Result<Option<Option<PropertyValue>>, GraphError> {
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
let Some(guard) = nodes.get(id.0)? else {
return Ok(None);
};
match crate::encode::node_prop_raw(guard.value(), prop_id)? {
Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
None => Ok(Some(None)),
}
}
pub fn get_edge_prop_in_txn(
txn: Txn,
id: EdgeId,
prop_id: u32,
) -> Result<Option<Option<PropertyValue>>, GraphError> {
let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
let Some(guard) = edges.get(id.0)? else {
return Ok(None);
};
match crate::encode::edge_prop_raw(guard.value(), prop_id)? {
Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
None => Ok(Some(None)),
}
}
#[allow(clippy::type_complexity)] pub fn node_prop_reader(
txn: Txn<'_>,
) -> Result<
impl FnMut(NodeId, u32) -> Result<Option<Option<PropertyValue>>, GraphError> + '_,
GraphError,
> {
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
Ok(move |id: NodeId, prop_id: u32| {
let Some(guard) = nodes.get(id.0)? else {
return Ok(None);
};
match crate::encode::node_prop_raw(guard.value(), prop_id)? {
Some(raw) => Ok(Some(Some(crate::encode::decode_value(raw)?))),
None => Ok(Some(None)),
}
})
}
pub fn node_exists_in_txn(txn: Txn, id: NodeId) -> Result<bool, GraphError> {
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
let exists = nodes.get(id.0)?.is_some();
Ok(exists)
}
pub fn edge_exists_in_txn(txn: Txn, id: EdgeId) -> Result<bool, GraphError> {
let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
let exists = edges.get(id.0)?.is_some();
Ok(exists)
}
pub fn create_edge(
&self,
label: &str,
src: NodeId,
dst: NodeId,
props: BTreeMap<String, PropertyValue>,
) -> Result<EdgeId, GraphError> {
let write_txn = self.begin_write()?;
let id = Self::create_edge_in_txn(&write_txn, label, src, dst, props)?;
write_txn.commit()?;
Ok(id)
}
pub fn create_edge_in_txn(
write_txn: &WriteTransaction,
label: &str,
src: NodeId,
dst: NodeId,
props: BTreeMap<String, PropertyValue>,
) -> Result<EdgeId, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
Self::create_edge_ctx(&mut ctx, label, src, dst, props)
}
fn create_edge_ctx(
ctx: &mut WriteCtx,
label: &str,
src: NodeId,
dst: NodeId,
props: BTreeMap<String, PropertyValue>,
) -> Result<EdgeId, GraphError> {
if ctx.nodes()?.get(src.0)?.is_none() {
return Err(GraphError::NodeNotFound(src));
}
if ctx.nodes()?.get(dst.0)?.is_none() {
return Err(GraphError::NodeNotFound(dst));
}
let label_id = intern_label(ctx, label)?;
let id = next_id(ctx, "next_edge_id")?;
let record = EdgeRecord {
label_id,
src: src.0,
dst: dst.0,
props,
};
let bytes = encode_edge(&record, |name| intern_prop(ctx, name))?;
ctx.edges()?.insert(id, bytes.as_slice())?;
ctx.adj_out()?
.insert(crate::model::adj_key(src.0, label_id, id), dst.0)?;
ctx.adj_in()?
.insert(crate::model::adj_key(dst.0, label_id, id), src.0)?;
Ok(EdgeId(id))
}
pub fn get_edge(&self, id: EdgeId) -> Result<Option<Edge>, GraphError> {
let read_txn = self.begin_read()?;
Self::get_edge_in_txn(Txn::Read(&read_txn), id)
}
pub fn get_edge_in_txn(txn: Txn, id: EdgeId) -> Result<Option<Edge>, GraphError> {
let record: Option<EdgeRecord> = {
let edges = txn.open_table(marsdb_storage::tables::EDGES)?;
let found = match edges.get(id.0)? {
Some(guard) => {
let mut resolve = prop_resolver(txn)?;
Some(decode_edge(guard.value(), &mut resolve)?)
}
None => None,
};
found
};
let Some(record) = record else {
return Ok(None);
};
let label = resolve_label(txn, record.label_id)?;
Ok(Some(Edge {
id,
label,
src: NodeId(record.src),
dst: NodeId(record.dst),
props: record.props,
}))
}
pub fn neighbors(
&self,
node: NodeId,
dir: Direction,
label_filter: Option<&str>,
) -> Result<Vec<AdjEntry>, GraphError> {
let read_txn = self.begin_read()?;
Self::neighbors_in_txn(Txn::Read(&read_txn), node, dir, label_filter)
}
pub fn neighbors_in_txn(
txn: Txn,
node: NodeId,
dir: Direction,
label_filter: Option<&str>,
) -> Result<Vec<AdjEntry>, GraphError> {
let (lo, hi) = match label_filter {
Some(l) => match lookup_label_id(txn, l)? {
Some(lid) => crate::model::adj_label_bounds(node.0, lid),
None => return Ok(Vec::new()),
},
None => crate::model::adj_node_bounds(node.0),
};
let mut result = Vec::new();
let table_def = match dir {
Direction::Out => marsdb_storage::tables::ADJ_OUT,
Direction::In => marsdb_storage::tables::ADJ_IN,
};
let table = txn.open_table(table_def)?;
for item in table.range(lo..=hi)? {
let (key, value) = item?;
let (_, label_id, edge_id) = key.value();
result.push(AdjEntry {
edge_id: EdgeId(edge_id),
other: NodeId(value.value()),
label_id,
});
}
Ok(result)
}
pub fn delete_edge(&self, id: EdgeId) -> Result<bool, GraphError> {
let write_txn = self.begin_write()?;
let removed = Self::delete_edge_in_txn(&write_txn, id)?;
write_txn.commit()?;
Ok(removed)
}
pub fn delete_edge_in_txn(
write_txn: &WriteTransaction,
id: EdgeId,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
Self::delete_edge_ctx(&mut ctx, id)
}
fn delete_edge_ctx(ctx: &mut WriteCtx, id: EdgeId) -> Result<bool, GraphError> {
let Some(record_bytes) = ctx
.edges()?
.remove(id.0)?
.map(|guard| guard.value().to_vec())
else {
return Ok(false);
};
let (label_id, src, dst) = edge_header(&record_bytes)?;
ctx.adj_out()?
.remove(crate::model::adj_key(src, label_id, id.0))?;
ctx.adj_in()?
.remove(crate::model::adj_key(dst, label_id, id.0))?;
Ok(true)
}
pub fn delete_node(&self, id: NodeId, detach: bool) -> Result<bool, GraphError> {
let write_txn = self.begin_write()?;
let existed = Self::delete_node_in_txn(&write_txn, id, detach)?;
write_txn.commit()?;
Ok(existed)
}
pub fn delete_node_in_txn(
write_txn: &WriteTransaction,
id: NodeId,
detach: bool,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
let mut incident: Vec<EdgeId> = Vec::new();
let (lo, hi) = crate::model::adj_node_bounds(id.0);
for item in ctx.adj_out()?.range(lo..=hi)? {
let (key, _) = item?;
let (_, _, edge_id) = key.value();
incident.push(EdgeId(edge_id));
}
for item in ctx.adj_in()?.range(lo..=hi)? {
let (key, _) = item?;
let (_, _, edge_id) = key.value();
incident.push(EdgeId(edge_id));
}
if !incident.is_empty() && !detach {
return Err(GraphError::NodeHasEdges(id));
}
for edge_id in incident {
Self::delete_edge_ctx(&mut ctx, edge_id)?;
}
let Some(removed_bytes) = ctx
.nodes()?
.remove(id.0)?
.map(|guard| guard.value().to_vec())
else {
return Ok(false);
};
let record = decode_node(&removed_bytes, |pid| {
crate::index::resolve_prop_ctx(&mut ctx, pid)
})?;
for &label_id in &record.label_ids {
ctx.node_label_index()?.remove(label_id, id.0)?;
}
crate::index::on_node_deleted(&mut ctx, id.0, &record.label_ids, &record.props)?;
Ok(true)
}
pub fn create_index(&self, label: &str, prop: &str, unique: bool) -> Result<(), GraphError> {
let write_txn = self.begin_write()?;
Self::create_index_in_txn(&write_txn, label, prop, unique)?;
write_txn.commit()?;
Ok(())
}
pub fn create_index_in_txn(
write_txn: &WriteTransaction,
label: &str,
prop: &str,
unique: bool,
) -> Result<(), GraphError> {
let mut ctx = WriteCtx::open(write_txn);
crate::index::create_index(&mut ctx, label, prop, unique)
}
pub fn index_def(
&self,
label: &str,
prop: &str,
) -> Result<Option<crate::IndexDef>, GraphError> {
let read_txn = self.begin_read()?;
crate::index::lookup_index_def(Txn::Read(&read_txn), label, prop)
}
pub fn index_def_in_txn(
txn: Txn,
label: &str,
prop: &str,
) -> Result<Option<crate::IndexDef>, GraphError> {
crate::index::lookup_index_def(txn, label, prop)
}
pub fn lookup_by_index_in_txn(
txn: Txn,
label: &str,
prop: &str,
value: &PropertyValue,
) -> Result<Vec<NodeId>, GraphError> {
crate::index::lookup_exact(txn, label, prop, value, None)
}
pub fn lookup_by_index_limited_in_txn(
txn: Txn,
label: &str,
prop: &str,
value: &PropertyValue,
limit: usize,
) -> Result<Vec<NodeId>, GraphError> {
crate::index::lookup_exact(txn, label, prop, value, Some(limit))
}
pub fn index_match_count_in_txn(
txn: Txn,
label: &str,
prop: &str,
value: &PropertyValue,
) -> Result<u64, GraphError> {
crate::index::match_count(txn, label, prop, value)
}
pub fn lookup_by_index(
&self,
label: &str,
prop: &str,
value: &PropertyValue,
) -> Result<Vec<NodeId>, GraphError> {
let read_txn = self.begin_read()?;
crate::index::lookup_exact(Txn::Read(&read_txn), label, prop, value, None)
}
pub fn set_node_prop(
&self,
id: NodeId,
key: &str,
value: PropertyValue,
) -> Result<bool, GraphError> {
let write_txn = self.begin_write()?;
let updated = Self::set_node_prop_in_txn(&write_txn, id, key, value)?;
if updated {
write_txn.commit()?;
} else {
write_txn.abort()?;
}
Ok(updated)
}
pub fn set_node_prop_in_txn(
write_txn: &WriteTransaction,
id: NodeId,
key: &str,
value: PropertyValue,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
return Ok(false);
};
let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
let old_value = record.props.insert(key.to_string(), value.clone());
let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
crate::index::on_node_prop_changed(
&mut ctx,
id.0,
&record.label_ids,
key,
old_value.as_ref(),
Some(&value),
)?;
Ok(true)
}
pub fn set_edge_prop(
&self,
id: EdgeId,
key: &str,
value: PropertyValue,
) -> Result<bool, GraphError> {
let write_txn = self.begin_write()?;
let updated = Self::set_edge_prop_in_txn(&write_txn, id, key, value)?;
if updated {
write_txn.commit()?;
} else {
write_txn.abort()?;
}
Ok(updated)
}
pub fn set_edge_prop_in_txn(
write_txn: &WriteTransaction,
id: EdgeId,
key: &str,
value: PropertyValue,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
let Some(bytes) = ctx.edges()?.get(id.0)?.map(|g| g.value().to_vec()) else {
return Ok(false);
};
let mut record = decode_edge(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
record.props.insert(key.to_string(), value);
let new_bytes = encode_edge(&record, |name| intern_prop(&mut ctx, name))?;
ctx.edges()?.insert(id.0, new_bytes.as_slice())?;
Ok(true)
}
pub fn remove_node_prop_in_txn(
write_txn: &WriteTransaction,
id: NodeId,
key: &str,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
return Ok(false);
};
let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
let old_value = record.props.remove(key);
let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
crate::index::on_node_prop_changed(
&mut ctx,
id.0,
&record.label_ids,
key,
old_value.as_ref(),
None,
)?;
Ok(true)
}
pub fn remove_edge_prop_in_txn(
write_txn: &WriteTransaction,
id: EdgeId,
key: &str,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
let Some(bytes) = ctx.edges()?.get(id.0)?.map(|g| g.value().to_vec()) else {
return Ok(false);
};
let mut record = decode_edge(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
record.props.remove(key);
let new_bytes = encode_edge(&record, |name| intern_prop(&mut ctx, name))?;
ctx.edges()?.insert(id.0, new_bytes.as_slice())?;
Ok(true)
}
pub fn add_node_label_in_txn(
write_txn: &WriteTransaction,
id: NodeId,
label: &str,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
return Ok(false);
};
let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
let label_id = intern_label(&mut ctx, label)?;
if !record.label_ids.contains(&label_id) {
record.label_ids.push(label_id);
let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
ctx.node_label_index()?.insert(label_id, id.0)?;
crate::index::on_node_created(&mut ctx, id.0, &[label_id], &record.props)?;
}
Ok(true)
}
pub fn remove_node_label_in_txn(
write_txn: &WriteTransaction,
id: NodeId,
label: &str,
) -> Result<bool, GraphError> {
let mut ctx = WriteCtx::open(write_txn);
let Some(bytes) = ctx.nodes()?.get(id.0)?.map(|g| g.value().to_vec()) else {
return Ok(false);
};
let Some(label_id) = ctx.label_to_id()?.get(label)?.map(|g| g.value()) else {
return Ok(true);
};
let mut record = decode_node(&bytes, |pid| crate::index::resolve_prop_ctx(&mut ctx, pid))?;
if let Some(pos) = record.label_ids.iter().position(|&l| l == label_id) {
record.label_ids.remove(pos);
let new_bytes = encode_node(&record, |name| intern_prop(&mut ctx, name))?;
ctx.nodes()?.insert(id.0, new_bytes.as_slice())?;
ctx.node_label_index()?.remove(label_id, id.0)?;
crate::index::on_node_deleted(&mut ctx, id.0, &[label_id], &record.props)?;
}
Ok(true)
}
pub fn node_count_in_txn(txn: Txn) -> Result<u64, GraphError> {
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
Ok(nodes.len()?)
}
pub fn label_count_in_txn(txn: Txn, label: &str) -> Result<u64, GraphError> {
let Some(label_id) = lookup_label_id(txn, label)? else {
return Ok(0);
};
let index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
let count = index.get(label_id)?.len();
Ok(count)
}
pub fn all_nodes(&self, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
let read_txn = self.begin_read()?;
Self::all_nodes_in_txn(Txn::Read(&read_txn), label_filter)
}
pub fn all_node_ids_limited_in_txn(
txn: Txn,
label_filter: Option<&str>,
limit: usize,
) -> Result<Vec<NodeId>, GraphError> {
let Some(label_filter) = label_filter else {
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
return nodes
.iter()?
.take(limit)
.map(|entry| {
entry
.map(|(key, _)| NodeId(key.value()))
.map_err(Into::into)
})
.collect();
};
let Some(label_id) = lookup_label_id(txn, label_filter)? else {
return Ok(Vec::new());
};
let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
let ids = label_index
.get(label_id)?
.take(limit)
.map(|entry| entry.map(|value| NodeId(value.value())).map_err(Into::into))
.collect::<Result<Vec<_>, GraphError>>()?;
drop(label_index);
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
for id in &ids {
if nodes.get(id.0)?.is_none() {
return Err(GraphError::CorruptData(format!(
"node label index references missing node {}",
id.0
)));
}
}
Ok(ids)
}
pub fn all_nodes_in_txn(txn: Txn, label_filter: Option<&str>) -> Result<Vec<Node>, GraphError> {
Self::all_nodes_limited_in_txn(txn, label_filter, usize::MAX)
}
pub fn all_nodes_limited_in_txn(
txn: Txn,
label_filter: Option<&str>,
limit: usize,
) -> Result<Vec<Node>, GraphError> {
let Some(label_filter) = label_filter else {
let mut result = Vec::new();
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
let mut resolve = prop_resolver(txn)?;
for item in nodes.iter()? {
if result.len() >= limit {
break;
}
let (key, value) = item?;
let record = decode_node(value.value(), &mut resolve)?;
let labels = record
.label_ids
.iter()
.map(|&lid| resolve_label(txn, lid))
.collect::<Result<Vec<_>, _>>()?;
result.push(Node {
id: NodeId(key.value()),
labels,
props: record.props,
});
}
return Ok(result);
};
let Some(label_id) = lookup_label_id(txn, label_filter)? else {
return Ok(Vec::new());
};
let node_ids: Vec<u64> = {
let label_index = txn.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)?;
let ids: Vec<u64> = label_index
.get(label_id)?
.take(limit)
.map(|item| item.map(|g| g.value()))
.collect::<Result<_, _>>()?;
ids
};
let mut result = Vec::with_capacity(node_ids.len());
let nodes = txn.open_table(marsdb_storage::tables::NODES)?;
let mut resolve = prop_resolver(txn)?;
for id in node_ids {
let guard = nodes.get(id)?.ok_or_else(|| {
GraphError::CorruptData(format!("node label index references missing node {}", id))
})?;
let record = decode_node(guard.value(), &mut resolve)?;
drop(guard);
let labels = record
.label_ids
.iter()
.map(|&lid| resolve_label(txn, lid))
.collect::<Result<Vec<_>, _>>()?;
result.push(Node {
id: NodeId(id),
labels,
props: record.props,
});
}
Ok(result)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn integrity_check_rejects_missing_node_label_index_entry() {
let mut store = GraphStore::open_memory().unwrap();
let node = store.create_node(&["Person"], BTreeMap::new()).unwrap();
let write = store.begin_write().unwrap();
let label_id = {
let labels = write
.open_table(marsdb_storage::tables::LABEL_TO_ID)
.unwrap();
let id = labels.get("Person").unwrap().unwrap().value();
id
};
write
.open_multimap_table(marsdb_storage::tables::NODE_LABEL_INDEX)
.unwrap()
.remove(label_id, node.0)
.unwrap();
write.commit().unwrap();
let error = store.check_integrity().unwrap_err();
assert!(
matches!(error, GraphError::CorruptData(message) if message.contains("missing from the label index"))
);
}
#[test]
fn integrity_check_rejects_dangling_adjacency_entry() {
let mut store = GraphStore::open_memory().unwrap();
let node = store.create_node(&[], BTreeMap::new()).unwrap();
let write = store.begin_write().unwrap();
write
.open_table(marsdb_storage::tables::ADJ_OUT)
.unwrap()
.insert(crate::model::adj_key(node.0, 0, 999), node.0)
.unwrap();
write.commit().unwrap();
let error = store.check_integrity().unwrap_err();
assert!(
matches!(error, GraphError::CorruptData(message) if message.contains("missing edge 999"))
);
}
#[test]
fn create_index_backfills_existing_nodes() {
let store = GraphStore::open_memory().unwrap();
let mut alice_props = BTreeMap::new();
alice_props.insert(
"email".to_string(),
PropertyValue::String("alice@x.com".to_string()),
);
let alice = store.create_node(&["Person"], alice_props).unwrap();
let mut bob_props = BTreeMap::new();
bob_props.insert(
"email".to_string(),
PropertyValue::String("bob@x.com".to_string()),
);
store.create_node(&["Person"], bob_props).unwrap();
store.create_node(&["Person"], BTreeMap::new()).unwrap();
store.create_index("Person", "email", false).unwrap();
let found = store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("alice@x.com".to_string()),
)
.unwrap();
assert_eq!(found, vec![alice]);
}
#[test]
fn create_index_rejects_duplicate_unique_value() {
let store = GraphStore::open_memory().unwrap();
let mut props1 = BTreeMap::new();
props1.insert(
"email".to_string(),
PropertyValue::String("same@x.com".to_string()),
);
store.create_node(&["Person"], props1).unwrap();
let mut props2 = BTreeMap::new();
props2.insert(
"email".to_string(),
PropertyValue::String("same@x.com".to_string()),
);
store.create_node(&["Person"], props2).unwrap();
let error = store.create_index("Person", "email", true).unwrap_err();
assert!(matches!(
error,
GraphError::UniqueConstraintViolation { .. }
));
assert!(store.index_def("Person", "email").unwrap().is_none());
}
#[test]
fn lookup_by_index_on_undeclared_index_is_empty_not_an_error() {
let store = GraphStore::open_memory().unwrap();
store.create_node(&["Person"], BTreeMap::new()).unwrap();
let found = store
.lookup_by_index("Person", "email", &PropertyValue::String("x".to_string()))
.unwrap();
assert_eq!(found, Vec::new());
assert!(store.index_def("Person", "email").unwrap().is_none());
}
#[test]
fn index_survives_reopen() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("index.db");
{
let store = GraphStore::open_file(&path).unwrap();
let mut props = BTreeMap::new();
props.insert(
"email".to_string(),
PropertyValue::String("x@x.com".to_string()),
);
store.create_node(&["Person"], props).unwrap();
store.create_index("Person", "email", false).unwrap();
}
let store = GraphStore::open_file(&path).unwrap();
assert!(store.index_def("Person", "email").unwrap().is_some());
let found = store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("x@x.com".to_string()),
)
.unwrap();
assert_eq!(found.len(), 1);
}
#[test]
fn create_node_after_index_declared_is_indexed_immediately() {
let store = GraphStore::open_memory().unwrap();
store.create_index("Person", "email", false).unwrap();
let mut props = BTreeMap::new();
props.insert(
"email".to_string(),
PropertyValue::String("new@x.com".to_string()),
);
let node = store.create_node(&["Person"], props).unwrap();
let found = store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("new@x.com".to_string()),
)
.unwrap();
assert_eq!(found, vec![node]);
}
#[test]
fn set_node_prop_moves_the_index_entry() {
let store = GraphStore::open_memory().unwrap();
let mut props = BTreeMap::new();
props.insert(
"email".to_string(),
PropertyValue::String("old@x.com".to_string()),
);
let node = store.create_node(&["Person"], props).unwrap();
store.create_index("Person", "email", false).unwrap();
store
.set_node_prop(
node,
"email",
PropertyValue::String("new@x.com".to_string()),
)
.unwrap();
assert!(store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("old@x.com".to_string())
)
.unwrap()
.is_empty());
assert_eq!(
store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("new@x.com".to_string())
)
.unwrap(),
vec![node]
);
}
#[test]
fn set_node_prop_enforces_unique_index() {
let store = GraphStore::open_memory().unwrap();
let mut props1 = BTreeMap::new();
props1.insert(
"email".to_string(),
PropertyValue::String("a@x.com".to_string()),
);
store.create_node(&["Person"], props1).unwrap();
let mut props2 = BTreeMap::new();
props2.insert(
"email".to_string(),
PropertyValue::String("b@x.com".to_string()),
);
let node2 = store.create_node(&["Person"], props2).unwrap();
store.create_index("Person", "email", true).unwrap();
let error = store
.set_node_prop(node2, "email", PropertyValue::String("a@x.com".to_string()))
.unwrap_err();
assert!(matches!(
error,
GraphError::UniqueConstraintViolation { .. }
));
}
#[test]
fn remove_node_prop_removes_the_index_entry() {
let store = GraphStore::open_memory().unwrap();
let mut props = BTreeMap::new();
props.insert(
"email".to_string(),
PropertyValue::String("gone@x.com".to_string()),
);
let node = store.create_node(&["Person"], props).unwrap();
store.create_index("Person", "email", false).unwrap();
let write = store.begin_write().unwrap();
GraphStore::remove_node_prop_in_txn(&write, node, "email").unwrap();
write.commit().unwrap();
assert!(store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("gone@x.com".to_string())
)
.unwrap()
.is_empty());
}
#[test]
fn delete_node_removes_its_index_entries() {
let store = GraphStore::open_memory().unwrap();
let mut props = BTreeMap::new();
props.insert(
"email".to_string(),
PropertyValue::String("deleted@x.com".to_string()),
);
let node = store.create_node(&["Person"], props).unwrap();
store.create_index("Person", "email", false).unwrap();
store.delete_node(node, false).unwrap();
assert!(store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("deleted@x.com".to_string())
)
.unwrap()
.is_empty());
}
#[test]
fn add_node_label_indexes_existing_props_under_the_new_label() {
let store = GraphStore::open_memory().unwrap();
let mut props = BTreeMap::new();
props.insert(
"email".to_string(),
PropertyValue::String("multi@x.com".to_string()),
);
let node = store.create_node(&["Contact"], props).unwrap();
store.create_index("Person", "email", false).unwrap();
assert!(store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("multi@x.com".to_string())
)
.unwrap()
.is_empty());
let write = store.begin_write().unwrap();
GraphStore::add_node_label_in_txn(&write, node, "Person").unwrap();
write.commit().unwrap();
assert_eq!(
store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("multi@x.com".to_string())
)
.unwrap(),
vec![node]
);
}
#[test]
fn remove_node_label_removes_index_entries_under_that_label() {
let store = GraphStore::open_memory().unwrap();
let mut props = BTreeMap::new();
props.insert(
"email".to_string(),
PropertyValue::String("dual@x.com".to_string()),
);
let node = store.create_node(&["Person", "Contact"], props).unwrap();
store.create_index("Person", "email", false).unwrap();
assert_eq!(
store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("dual@x.com".to_string())
)
.unwrap(),
vec![node]
);
let write = store.begin_write().unwrap();
GraphStore::remove_node_label_in_txn(&write, node, "Person").unwrap();
write.commit().unwrap();
assert!(store
.lookup_by_index(
"Person",
"email",
&PropertyValue::String("dual@x.com".to_string())
)
.unwrap()
.is_empty());
}
}