use super::{
push_str, push_u64, read_str, read_u64, single_str_key, KeyValueBatch, KeyValueCatalog,
StorageBackendError, StorageBackendResult, TAG_PATH_INDEX,
};
use crate::key_value::view::for_each_key;
use crate::key_value::KeyValueRead;
use crate::key_value::TAG_PATH_INDEX_DATA;
use crate::MAX_GRAPH_ID_PAGE;
fn key(kind: u8, name: &str) -> StorageBackendResult<Vec<u8>> {
let mut key = vec![TAG_PATH_INDEX_DATA, kind];
push_str(&mut key, name)?;
Ok(key)
}
fn reverse_key(graph: &str, index: &str) -> StorageBackendResult<Vec<u8>> {
let mut key = key(b'g', graph)?;
push_str(&mut key, index)?;
Ok(key)
}
fn pair_prefix(index: &str, sequence: &str) -> StorageBackendResult<Vec<u8>> {
let mut key = key(b'p', index)?;
push_str(&mut key, sequence)?;
Ok(key)
}
fn pair_key(mut prefix: Vec<u8>, source: u64, target: u64) -> Vec<u8> {
push_u64(&mut prefix, source);
push_u64(&mut prefix, target);
prefix
}
impl KeyValueCatalog {
pub(super) fn invalidate_graph_path_data(
read: &dyn KeyValueRead,
batch: &mut dyn KeyValueBatch,
graph: &str,
) -> StorageBackendResult<()> {
batch.graph_mutation(crate::mvcc::GraphMutation::InvalidateGraph(graph))?;
let prefix = key(b'g', graph)?;
for_each_key(read, &prefix, &mut |stored_key| {
let mut offset = prefix.len();
let index = read_str(stored_key, &mut offset)?;
if read.get(&key(b's', &index)?)?.as_deref() == Some(graph.as_bytes()) {
batch.preview_graph_invalidation(&key(b'v', &index)?, None)?;
}
Ok(true)
})
}
pub(super) fn clear_path_index_data_into(
read: &dyn KeyValueRead,
batch: &mut dyn KeyValueBatch,
index: &str,
) -> StorageBackendResult<()> {
let state = key(b's', index)?;
if let Some(graph) = read.get(&state)? {
let graph = std::str::from_utf8(&graph)
.map_err(|error| StorageBackendError::Other(error.to_string()))?;
batch.delete(&reverse_key(graph, index)?)?;
}
batch.delete_prefix(&key(b'p', index)?)?;
batch.replace_graph_cache(&key(b'v', index)?, None)?;
batch.delete(&state)
}
pub(super) fn clear_path_index_data_impl(&self, index: &str) -> StorageBackendResult<()> {
self.store
.with_mutation(&mut |read, batch| Self::clear_path_index_data_into(read, batch, index))
}
pub(super) fn invalidate_path_index_data_into(
batch: &mut dyn KeyValueBatch,
index: &str,
) -> StorageBackendResult<()> {
batch.replace_graph_cache(&key(b'v', index)?, None)
}
pub(super) fn save_path_index_pairs_impl(
&self,
index: &str,
sequence: &str,
pairs: &[(u64, u64)],
) -> StorageBackendResult<()> {
if pairs.len() > MAX_GRAPH_ID_PAGE {
return Err(StorageBackendError::Other(
"path index batch exceeds the bounded page size".into(),
));
}
let prefix = pair_prefix(index, sequence)?;
let mut batch = self.store.batch();
for &(source, target) in pairs {
batch.put(&pair_key(prefix.clone(), source, target), &[])?;
}
batch.commit()
}
pub(super) fn finish_path_index_data_impl(
&self,
index: &str,
graph: &str,
definition: &str,
) -> StorageBackendResult<()> {
self.store.with_mutation(&mut |read, batch| {
if read
.get(&single_str_key(TAG_PATH_INDEX, index)?)?
.as_deref()
!= Some(definition.as_bytes())
{
return Err(StorageBackendError::Other(format!(
"path index {index:?} definition changed during build"
)));
}
if let Some(previous) = read.get(&key(b's', index)?)? {
let previous = std::str::from_utf8(&previous)
.map_err(|error| StorageBackendError::Other(error.to_string()))?;
if previous != graph {
batch.delete(&reverse_key(previous, index)?)?;
}
}
batch.put(&key(b's', index)?, graph.as_bytes())?;
batch.replace_graph_cache(&key(b'v', index)?, Some(definition.as_bytes()))?;
batch.put(&reverse_key(graph, index)?, &[])?;
batch.graph_mutation(crate::mvcc::GraphMutation::PublishPath {
index,
graph,
definition,
})?;
Ok(())
})
}
pub(super) fn path_index_data_is_current_impl(
&self,
index: &str,
definition: &str,
) -> StorageBackendResult<bool> {
Ok(self.store.get(&key(b'v', index)?)?.as_deref() == Some(definition.as_bytes()))
}
pub(super) fn path_index_pairs_impl(
&self,
index: &str,
sequence: &str,
after: Option<(u64, u64)>,
limit: usize,
) -> StorageBackendResult<Vec<(u64, u64)>> {
crate::catalog::validate_graph_page(limit)?;
let prefix = pair_prefix(index, sequence)?;
let after = after.map(|(source, target)| pair_key(prefix.clone(), source, target));
self.store
.scan_prefix_keys_after(&prefix, after.as_deref(), limit)?
.into_iter()
.map(|key| {
let mut offset = prefix.len();
let source = read_u64(&key, &mut offset)?;
let target = read_u64(&key, &mut offset)?;
if offset != key.len() {
return Err(StorageBackendError::Other(
"malformed durable path-index pair key".into(),
));
}
Ok((source, target))
})
.collect()
}
}