use super::{ProjectedGraphArtifact, ProjectedGraphArtifactData, ProjectedGraphDefinition};
use crate::text::{
decode_string, decode_string_vec, decode_u64_vec, encode_string, encode_string_vec,
encode_u64_vec, parse_u64,
};
use crate::NodeId;
use hawdb_core::{HawDBError, Result};
use std::collections::BTreeMap;
const PROJECTED_GRAPH_ARTIFACT_VERSION: u64 = 2;
pub fn encode_projected_graph_artifacts<'a>(
projection_epoch: u64,
commit_epoch: u64,
artifacts: impl IntoIterator<
Item = (
&'a str,
&'a ProjectedGraphDefinition,
ProjectedGraphArtifactData,
),
>,
) -> String {
let mut body = String::new();
body.push_str("HAWDB_PROJECTED_GRAPHS_V1\n");
body.push_str(&format!(
"artifact_version\t{PROJECTED_GRAPH_ARTIFACT_VERSION}\n"
));
body.push_str(&format!("projection_epoch\t{projection_epoch}\n"));
body.push_str(&format!("commit_epoch\t{commit_epoch}\n"));
for (name, definition, data) in artifacts {
body.push_str(&format!(
"graph\t{}\t{}\t{}\t{}\t{}\t{}\n",
encode_string(name),
encode_string_vec(&definition.node_labels),
encode_string_vec(&definition.rel_types),
super::encode_projected_relationship_predicates(&definition.relationship_predicates),
data.node_count(),
data.edge_count()
));
body.push_str(&format!(
"nodes\t{}\n",
encode_u64_vec(data.nodes.iter().map(|node| node.0))
));
body.push_str(&format!(
"csr_offsets\t{}\n",
encode_usize_vec(data.csr_offsets.iter().copied())
));
body.push_str(&format!(
"csr_targets\t{}\n",
encode_usize_vec(data.csr_targets.iter().copied())
));
body.push_str(&format!(
"csc_offsets\t{}\n",
encode_usize_vec(data.csc_offsets.iter().copied())
));
body.push_str(&format!(
"csc_sources\t{}\n",
encode_usize_vec(data.csc_sources.iter().copied())
));
}
body
}
pub fn decode_projected_graph_artifacts(
body: &str,
) -> Result<(u64, BTreeMap<String, ProjectedGraphArtifact>)> {
let mut lines = body.lines();
match lines.next() {
Some("HAWDB_PROJECTED_GRAPHS_V1") => {}
_ => {
return Err(HawDBError::Storage(
"invalid projected graph artifact header".to_string(),
));
}
}
let artifact_version = decode_projected_graph_u64_header(
lines.next(),
"artifact_version",
"projected graph artifact version",
)?;
if !matches!(artifact_version, 1 | PROJECTED_GRAPH_ARTIFACT_VERSION) {
return Err(HawDBError::Storage(format!(
"unsupported projected graph artifact version: {artifact_version}"
)));
}
let projection_epoch = decode_projected_graph_u64_header(
lines.next(),
"projection_epoch",
"projected graph artifact projection epoch",
)?;
let commit_epoch = decode_projected_graph_u64_header(
lines.next(),
"commit_epoch",
"projected graph artifact commit epoch",
)?;
let mut artifacts = BTreeMap::new();
while let Some(line) = lines.next() {
let fields = line.split('\t').collect::<Vec<_>>();
let graph_fields = match (artifact_version, fields.as_slice()) {
(
1,
["graph", raw_name, raw_node_labels, raw_rel_types, raw_node_count, raw_edge_count],
) => Some((
*raw_name,
*raw_node_labels,
*raw_rel_types,
None,
*raw_node_count,
*raw_edge_count,
)),
(
PROJECTED_GRAPH_ARTIFACT_VERSION,
["graph", raw_name, raw_node_labels, raw_rel_types, raw_relationship_predicates, raw_node_count, raw_edge_count],
) => Some((
*raw_name,
*raw_node_labels,
*raw_rel_types,
Some(*raw_relationship_predicates),
*raw_node_count,
*raw_edge_count,
)),
_ => None,
};
match graph_fields {
Some((
raw_name,
raw_node_labels,
raw_rel_types,
raw_relationship_predicates,
raw_node_count,
raw_edge_count,
)) => {
let name = decode_string(raw_name)?;
let definition = ProjectedGraphDefinition {
node_labels: decode_string_vec(raw_node_labels)?,
rel_types: decode_string_vec(raw_rel_types)?,
relationship_predicates: raw_relationship_predicates
.map(super::decode_projected_relationship_predicates)
.transpose()?
.unwrap_or_default(),
};
let node_count = parse_u64(raw_node_count, "projected graph artifact node count")?;
let edge_count = parse_u64(raw_edge_count, "projected graph artifact edge count")?;
let nodes = decode_projected_graph_nodes_line(lines.next())?;
let csr_offsets = decode_projected_graph_usize_line(lines.next(), "csr_offsets")?;
let csr_targets = decode_projected_graph_usize_line(lines.next(), "csr_targets")?;
let csc_offsets = decode_projected_graph_usize_line(lines.next(), "csc_offsets")?;
let csc_sources = decode_projected_graph_usize_line(lines.next(), "csc_sources")?;
if nodes.len() as u64 != node_count {
return Err(HawDBError::Storage(format!(
"projected graph artifact node count mismatch for {name}"
)));
}
if csr_targets.len() as u64 != edge_count || csc_sources.len() as u64 != edge_count
{
return Err(HawDBError::Storage(format!(
"projected graph artifact edge count mismatch for {name}"
)));
}
let data = ProjectedGraphArtifactData::new(
nodes,
csr_offsets,
csr_targets,
csc_offsets,
csc_sources,
)
.map_err(HawDBError::Storage)?;
artifacts.insert(
name,
ProjectedGraphArtifact {
projection_epoch,
commit_epoch,
definition,
data,
},
);
}
None if fields.as_slice() == [""] => {}
_ => {
return Err(HawDBError::Storage(format!(
"invalid projected graph artifact line: {line}"
)));
}
}
}
Ok((commit_epoch, artifacts))
}
fn decode_projected_graph_u64_header(
line: Option<&str>,
expected: &str,
name: &str,
) -> Result<u64> {
let Some(line) = line else {
return Err(HawDBError::Storage(format!(
"missing projected graph artifact {expected}"
)));
};
let fields = line.split('\t').collect::<Vec<_>>();
match fields.as_slice() {
[field, raw] if *field == expected => parse_u64(raw, name),
_ => Err(HawDBError::Storage(format!(
"invalid projected graph artifact line: {line}"
))),
}
}
fn decode_projected_graph_nodes_line(line: Option<&str>) -> Result<Vec<NodeId>> {
let Some(line) = line else {
return Err(HawDBError::Storage(
"missing projected graph artifact nodes line".to_string(),
));
};
let fields = line.split('\t').collect::<Vec<_>>();
match fields.as_slice() {
["nodes", raw_values] => decode_u64_vec(raw_values, "projected graph artifact node id")
.map(|nodes| nodes.into_iter().map(NodeId).collect()),
_ => Err(HawDBError::Storage(format!(
"invalid projected graph artifact line: {line}"
))),
}
}
fn decode_projected_graph_usize_line(line: Option<&str>, expected: &str) -> Result<Vec<usize>> {
let Some(line) = line else {
return Err(HawDBError::Storage(format!(
"missing projected graph artifact {expected} line"
)));
};
let fields = line.split('\t').collect::<Vec<_>>();
match fields.as_slice() {
[name, raw_values] if *name == expected => {
decode_usize_vec(raw_values, "projected graph artifact index")
}
_ => Err(HawDBError::Storage(format!(
"invalid projected graph artifact line: {line}"
))),
}
}
pub fn split_projected_graph_artifact_checksum(text: &str) -> Result<(&str, u64)> {
let Some((body, footer)) = text.rsplit_once("checksum\t") else {
return Err(HawDBError::Storage(
"projected graph artifact missing checksum footer".to_string(),
));
};
let checksum = parse_u64(footer.trim(), "projected graph artifact checksum")?;
Ok((body, checksum))
}
fn encode_usize_vec(values: impl IntoIterator<Item = usize>) -> String {
values
.into_iter()
.map(|value| value.to_string())
.collect::<Vec<_>>()
.join(",")
}
fn decode_usize_vec(input: &str, name: &str) -> Result<Vec<usize>> {
if input.is_empty() {
return Ok(Vec::new());
}
input
.split(',')
.map(|value| {
value
.parse()
.map_err(|_| HawDBError::Storage(format!("invalid {name}: {value}")))
})
.collect()
}
#[cfg(test)]
mod tests;