use crate::cow::NodeRecordRead;
use crate::{
adjacency::{
AdjacencyDirection, AdjacencyGroupConsistencyMismatch, AdjacencyGroupKey, AdjacencyLayout,
AdjacencyPostingList,
},
cow::{CowSegment, CowSegmentedMap},
NodeId, RelId, RelRecord,
};
use hawdb_core::{BasicGraphStatistics, LabelId, RelTypeId, Value};
use std::collections::{BTreeMap, BTreeSet};
pub const DENSE_ADJACENCY_DEGREE_THRESHOLD: usize = 64;
const MAX_ADJACENCY_CONSISTENCY_SAMPLES: usize = 32;
type AdjacencyGroups = BTreeMap<AdjacencyGroupKey, BTreeSet<RelId>>;
type NodePropertyIndex = CowSegmentedMap<(LabelId, String, Value), CowSegment<BTreeSet<NodeId>>>;
type RelationshipPropertyIndex =
CowSegmentedMap<(RelTypeId, String, Value), CowSegment<BTreeSet<RelId>>>;
pub fn adjacency_layout_for_degree(degree: usize) -> AdjacencyLayout {
if degree >= DENSE_ADJACENCY_DEGREE_THRESHOLD {
AdjacencyLayout::Dense
} else {
AdjacencyLayout::Sparse
}
}
pub fn adjacency_consolidation_plan(
candidates: &[AdjacencyConsolidationCandidate],
) -> AdjacencyConsolidationPlan {
candidates.iter().fold(
AdjacencyConsolidationPlan::default(),
|mut plan, candidate| {
plan.group_count = plan.group_count.saturating_add(1);
plan.delta_entry_count = plan
.delta_entry_count
.saturating_add(candidate.delta_entry_count);
plan.estimated_entries = plan
.estimated_entries
.saturating_add(candidate.estimated_entries);
plan
},
)
}
pub fn maintained_adjacency_groups(
outgoing: &CowSegmentedMap<(NodeId, RelTypeId), AdjacencyPostingList>,
incoming: &CowSegmentedMap<(NodeId, RelTypeId), AdjacencyPostingList>,
) -> AdjacencyGroups {
let mut groups = AdjacencyGroups::new();
for ((node_id, rel_type), rel_ids) in outgoing.iter() {
groups.insert(
AdjacencyGroupKey {
node_id: *node_id,
rel_type: *rel_type,
direction: AdjacencyDirection::Outgoing,
},
rel_ids
.iter_copied()
.map(|entry| entry.relationship_id)
.collect(),
);
}
for ((node_id, rel_type), rel_ids) in incoming.iter() {
groups.insert(
AdjacencyGroupKey {
node_id: *node_id,
rel_type: *rel_type,
direction: AdjacencyDirection::Incoming,
},
rel_ids
.iter_copied()
.map(|entry| entry.relationship_id)
.collect(),
);
}
groups
}
pub fn recompute_adjacency_groups(
relationships: &CowSegmentedMap<RelId, RelRecord>,
) -> AdjacencyGroups {
let mut groups = AdjacencyGroups::new();
for relationship in relationships.values() {
groups
.entry(AdjacencyGroupKey {
node_id: relationship.source,
rel_type: relationship.rel_type,
direction: AdjacencyDirection::Outgoing,
})
.or_default()
.insert(relationship.id);
groups
.entry(AdjacencyGroupKey {
node_id: relationship.target,
rel_type: relationship.rel_type,
direction: AdjacencyDirection::Incoming,
})
.or_default()
.insert(relationship.id);
}
groups
}
pub fn compute_degree_statistics_from_adjacency(
nodes: &impl NodeRecordRead,
outgoing: &CowSegmentedMap<(NodeId, RelTypeId), AdjacencyPostingList>,
incoming: &CowSegmentedMap<(NodeId, RelTypeId), AdjacencyPostingList>,
) -> BTreeMap<DegreeStatisticsKey, DegreeStatisticsEntry> {
compute_degree_statistics_from_groups(nodes, maintained_adjacency_groups(outgoing, incoming))
}
pub fn compute_degree_statistics_from_relationships(
nodes: &impl NodeRecordRead,
relationships: &CowSegmentedMap<RelId, RelRecord>,
) -> BTreeMap<DegreeStatisticsKey, DegreeStatisticsEntry> {
compute_degree_statistics_from_groups(nodes, recompute_adjacency_groups(relationships))
}
fn compute_degree_statistics_from_groups(
nodes: &impl NodeRecordRead,
groups: AdjacencyGroups,
) -> BTreeMap<DegreeStatisticsKey, DegreeStatisticsEntry> {
let rel_types = groups
.keys()
.map(|key| key.rel_type)
.collect::<BTreeSet<_>>();
let label_counts = label_counts_for_degree_statistics(nodes);
let mut statistics = BTreeMap::<DegreeStatisticsKey, DegreeStatisticsEntry>::new();
for (label_id, node_count) in &label_counts {
for rel_type in &rel_types {
for direction in [AdjacencyDirection::Outgoing, AdjacencyDirection::Incoming] {
statistics.insert(
DegreeStatisticsKey {
label_id: *label_id,
rel_type: *rel_type,
direction,
},
DegreeStatisticsEntry {
node_count: *node_count,
non_zero_node_count: 0,
relationship_count: 0,
max_degree: 0,
dense_node_count: 0,
},
);
}
}
}
for (group, rel_ids) in groups {
let Some(node) = nodes.get(&group.node_id) else {
continue;
};
let degree = rel_ids.len() as u64;
for label_id in &node.labels {
let entry = statistics
.entry(DegreeStatisticsKey {
label_id: *label_id,
rel_type: group.rel_type,
direction: group.direction,
})
.or_insert(DegreeStatisticsEntry {
node_count: label_counts.get(label_id).copied().unwrap_or_default(),
non_zero_node_count: 0,
relationship_count: 0,
max_degree: 0,
dense_node_count: 0,
});
entry.non_zero_node_count += 1;
entry.relationship_count += degree;
entry.max_degree = entry.max_degree.max(degree);
if rel_ids.len() >= DENSE_ADJACENCY_DEGREE_THRESHOLD {
entry.dense_node_count += 1;
}
}
}
statistics
}
fn label_counts_for_degree_statistics(nodes: &impl NodeRecordRead) -> BTreeMap<LabelId, u64> {
let mut label_counts = BTreeMap::new();
for node in nodes.values() {
for label_id in &node.labels {
*label_counts.entry(*label_id).or_default() += 1;
}
}
label_counts
}
pub fn adjacency_direction_sort_key(direction: AdjacencyDirection) -> u8 {
match direction {
AdjacencyDirection::Outgoing => 0,
AdjacencyDirection::Incoming => 1,
}
}
fn sample_relationship_ids(rel_ids: &BTreeSet<RelId>) -> Vec<RelId> {
rel_ids
.iter()
.copied()
.take(MAX_ADJACENCY_CONSISTENCY_SAMPLES)
.collect()
}
fn node_property_index_reference_count(index: &NodePropertyIndex) -> usize {
index.values().map(|node_ids| node_ids.len()).sum()
}
fn relationship_property_index_reference_count(index: &RelationshipPropertyIndex) -> usize {
index.values().map(|rel_ids| rel_ids.len()).sum()
}
fn property_index_mismatch_summary(
maintained: &NodePropertyIndex,
recomputed: &NodePropertyIndex,
) -> (usize, usize, usize, Vec<(LabelId, String, Value)>) {
let mut missing_key_count = 0usize;
let mut extra_key_count = 0usize;
let mut mismatched_key_count = 0usize;
let mut mismatched_keys = Vec::new();
for key in maintained
.keys()
.chain(recomputed.keys())
.cloned()
.collect::<BTreeSet<_>>()
{
match (maintained.get(&key), recomputed.get(&key)) {
(Some(left), Some(right)) if left == right => {}
(Some(_), Some(_)) => mismatched_key_count += 1,
(Some(_), None) => extra_key_count += 1,
(None, Some(_)) => missing_key_count += 1,
(None, None) => {}
}
if maintained.get(&key) != recomputed.get(&key)
&& mismatched_keys.len() < MAX_ADJACENCY_CONSISTENCY_SAMPLES
{
mismatched_keys.push(key);
}
}
(
missing_key_count,
extra_key_count,
mismatched_key_count,
mismatched_keys,
)
}
fn relationship_property_index_mismatch_summary(
maintained: &RelationshipPropertyIndex,
recomputed: &RelationshipPropertyIndex,
) -> (usize, usize, usize, Vec<(RelTypeId, String, Value)>) {
let mut missing_key_count = 0usize;
let mut extra_key_count = 0usize;
let mut mismatched_key_count = 0usize;
let mut mismatched_keys = Vec::new();
for key in maintained
.keys()
.chain(recomputed.keys())
.cloned()
.collect::<BTreeSet<_>>()
{
match (maintained.get(&key), recomputed.get(&key)) {
(Some(left), Some(right)) if left == right => {}
(Some(_), Some(_)) => mismatched_key_count += 1,
(Some(_), None) => extra_key_count += 1,
(None, Some(_)) => missing_key_count += 1,
(None, None) => {}
}
if maintained.get(&key) != recomputed.get(&key)
&& mismatched_keys.len() < MAX_ADJACENCY_CONSISTENCY_SAMPLES
{
mismatched_keys.push(key);
}
}
(
missing_key_count,
extra_key_count,
mismatched_key_count,
mismatched_keys,
)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdjacencyConsistencyReport {
pub ready: bool,
pub computed_at_commit_epoch: u64,
pub relationship_count: usize,
pub maintained_group_count: usize,
pub recomputed_group_count: usize,
pub dense_group_count: usize,
pub missing_group_count: usize,
pub extra_group_count: usize,
pub mismatched_group_count: usize,
pub dangling_relationship_count: usize,
pub mismatches: Vec<AdjacencyGroupConsistencyMismatch>,
pub dangling_relationship_ids: Vec<RelId>,
}
impl AdjacencyConsistencyReport {
pub fn new(
computed_at_commit_epoch: u64,
relationship_count: usize,
maintained: AdjacencyGroups,
recomputed: AdjacencyGroups,
relationships: &CowSegmentedMap<RelId, RelRecord>,
) -> Self {
let mut missing_group_count = 0;
let mut extra_group_count = 0;
let mut mismatched_group_count = 0;
let mut mismatches = Vec::new();
let keys = maintained
.keys()
.chain(recomputed.keys())
.copied()
.collect::<BTreeSet<_>>();
for key in keys {
let maintained_ids = maintained.get(&key);
let recomputed_ids = recomputed.get(&key);
if maintained_ids == recomputed_ids {
continue;
}
match (maintained_ids, recomputed_ids) {
(None, Some(_)) => missing_group_count += 1,
(Some(_), None) => extra_group_count += 1,
(Some(_), Some(_)) => mismatched_group_count += 1,
(None, None) => {}
}
if mismatches.len() < MAX_ADJACENCY_CONSISTENCY_SAMPLES {
mismatches.push(AdjacencyGroupConsistencyMismatch {
key,
maintained_relationship_ids: maintained_ids
.map(sample_relationship_ids)
.unwrap_or_default(),
recomputed_relationship_ids: recomputed_ids
.map(sample_relationship_ids)
.unwrap_or_default(),
});
}
}
let dangling_relationships = maintained
.values()
.flat_map(|rel_ids| rel_ids.iter().copied())
.filter(|rel_id| !relationships.contains_key(rel_id))
.collect::<BTreeSet<_>>();
let dangling_relationship_ids = dangling_relationships
.iter()
.copied()
.take(MAX_ADJACENCY_CONSISTENCY_SAMPLES)
.collect::<Vec<_>>();
let dangling_relationship_count = dangling_relationships.len();
let ready = missing_group_count == 0
&& extra_group_count == 0
&& mismatched_group_count == 0
&& dangling_relationship_count == 0;
Self {
ready,
computed_at_commit_epoch,
relationship_count,
maintained_group_count: maintained.len(),
recomputed_group_count: recomputed.len(),
dense_group_count: maintained
.values()
.filter(|rel_ids| rel_ids.len() >= DENSE_ADJACENCY_DEGREE_THRESHOLD)
.count(),
missing_group_count,
extra_group_count,
mismatched_group_count,
dangling_relationship_count,
mismatches,
dangling_relationship_ids,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct DegreeStatisticsKey {
pub label_id: LabelId,
pub rel_type: RelTypeId,
pub direction: AdjacencyDirection,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DegreeStatisticsEntry {
pub node_count: u64,
pub non_zero_node_count: u64,
pub relationship_count: u64,
pub max_degree: u64,
pub dense_node_count: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DegreeStatisticsConsistencyReport {
pub ready: bool,
pub computed_at_commit_epoch: u64,
pub maintained: BTreeMap<DegreeStatisticsKey, DegreeStatisticsEntry>,
pub recomputed: BTreeMap<DegreeStatisticsKey, DegreeStatisticsEntry>,
pub mismatched_keys: Vec<DegreeStatisticsKey>,
}
impl DegreeStatisticsConsistencyReport {
pub fn new(
computed_at_commit_epoch: u64,
maintained: BTreeMap<DegreeStatisticsKey, DegreeStatisticsEntry>,
recomputed: BTreeMap<DegreeStatisticsKey, DegreeStatisticsEntry>,
) -> Self {
let mismatched_keys = maintained
.keys()
.chain(recomputed.keys())
.copied()
.collect::<BTreeSet<_>>()
.into_iter()
.filter(|key| maintained.get(key) != recomputed.get(key))
.take(MAX_ADJACENCY_CONSISTENCY_SAMPLES)
.collect::<Vec<_>>();
Self {
ready: mismatched_keys.is_empty(),
computed_at_commit_epoch,
maintained,
recomputed,
mismatched_keys,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DistinctValueStatisticsConsistencyReport {
pub ready: bool,
pub computed_at_commit_epoch: u64,
pub maintained_property_distinct_counts: BTreeMap<(LabelId, String), u64>,
pub recomputed_property_distinct_counts: BTreeMap<(LabelId, String), u64>,
pub maintained_rel_property_distinct_counts: BTreeMap<(RelTypeId, String), u64>,
pub recomputed_rel_property_distinct_counts: BTreeMap<(RelTypeId, String), u64>,
pub mismatched_property_keys: Vec<(LabelId, String)>,
pub mismatched_rel_property_keys: Vec<(RelTypeId, String)>,
}
impl DistinctValueStatisticsConsistencyReport {
pub fn new(
computed_at_commit_epoch: u64,
maintained_property_distinct_counts: BTreeMap<(LabelId, String), u64>,
recomputed_property_distinct_counts: BTreeMap<(LabelId, String), u64>,
maintained_rel_property_distinct_counts: BTreeMap<(RelTypeId, String), u64>,
recomputed_rel_property_distinct_counts: BTreeMap<(RelTypeId, String), u64>,
) -> Self {
let mismatched_property_keys = maintained_property_distinct_counts
.keys()
.chain(recomputed_property_distinct_counts.keys())
.cloned()
.collect::<BTreeSet<_>>()
.into_iter()
.filter(|key| {
maintained_property_distinct_counts.get(key)
!= recomputed_property_distinct_counts.get(key)
})
.take(MAX_ADJACENCY_CONSISTENCY_SAMPLES)
.collect::<Vec<_>>();
let mismatched_rel_property_keys = maintained_rel_property_distinct_counts
.keys()
.chain(recomputed_rel_property_distinct_counts.keys())
.cloned()
.collect::<BTreeSet<_>>()
.into_iter()
.filter(|key| {
maintained_rel_property_distinct_counts.get(key)
!= recomputed_rel_property_distinct_counts.get(key)
})
.take(MAX_ADJACENCY_CONSISTENCY_SAMPLES)
.collect::<Vec<_>>();
Self {
ready: mismatched_property_keys.is_empty() && mismatched_rel_property_keys.is_empty(),
computed_at_commit_epoch,
maintained_property_distinct_counts,
recomputed_property_distinct_counts,
maintained_rel_property_distinct_counts,
recomputed_rel_property_distinct_counts,
mismatched_property_keys,
mismatched_rel_property_keys,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PropertyIndexConsistencyReport {
pub ready: bool,
pub computed_at_commit_epoch: u64,
pub node_index_entry_count: usize,
pub recomputed_node_index_entry_count: usize,
pub node_index_reference_count: usize,
pub recomputed_node_index_reference_count: usize,
pub missing_node_key_count: usize,
pub extra_node_key_count: usize,
pub mismatched_node_key_count: usize,
pub relationship_index_entry_count: usize,
pub recomputed_relationship_index_entry_count: usize,
pub relationship_index_reference_count: usize,
pub recomputed_relationship_index_reference_count: usize,
pub missing_relationship_key_count: usize,
pub extra_relationship_key_count: usize,
pub mismatched_relationship_key_count: usize,
pub mismatched_node_keys: Vec<(LabelId, String, Value)>,
pub mismatched_relationship_keys: Vec<(RelTypeId, String, Value)>,
}
impl PropertyIndexConsistencyReport {
pub fn new(
computed_at_commit_epoch: u64,
maintained_node_index: &NodePropertyIndex,
recomputed_node_index: &NodePropertyIndex,
maintained_relationship_index: &RelationshipPropertyIndex,
recomputed_relationship_index: &RelationshipPropertyIndex,
) -> Self {
let (
missing_node_key_count,
extra_node_key_count,
mismatched_node_key_count,
mismatched_node_keys,
) = property_index_mismatch_summary(maintained_node_index, recomputed_node_index);
let (
missing_relationship_key_count,
extra_relationship_key_count,
mismatched_relationship_key_count,
mismatched_relationship_keys,
) = relationship_property_index_mismatch_summary(
maintained_relationship_index,
recomputed_relationship_index,
);
Self {
ready: missing_node_key_count == 0
&& extra_node_key_count == 0
&& mismatched_node_key_count == 0
&& missing_relationship_key_count == 0
&& extra_relationship_key_count == 0
&& mismatched_relationship_key_count == 0,
computed_at_commit_epoch,
node_index_entry_count: maintained_node_index.len(),
recomputed_node_index_entry_count: recomputed_node_index.len(),
node_index_reference_count: node_property_index_reference_count(maintained_node_index),
recomputed_node_index_reference_count: node_property_index_reference_count(
recomputed_node_index,
),
missing_node_key_count,
extra_node_key_count,
mismatched_node_key_count,
relationship_index_entry_count: maintained_relationship_index.len(),
recomputed_relationship_index_entry_count: recomputed_relationship_index.len(),
relationship_index_reference_count: relationship_property_index_reference_count(
maintained_relationship_index,
),
recomputed_relationship_index_reference_count:
relationship_property_index_reference_count(recomputed_relationship_index),
missing_relationship_key_count,
extra_relationship_key_count,
mismatched_relationship_key_count,
mismatched_node_keys,
mismatched_relationship_keys,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BasicStatisticsConsistencyReport {
pub ready: bool,
pub computed_at_commit_epoch: u64,
pub incremental: BasicGraphStatistics,
pub recomputed: BasicGraphStatistics,
pub mismatched_fields: Vec<String>,
}
impl BasicStatisticsConsistencyReport {
pub fn new(incremental: BasicGraphStatistics, recomputed: BasicGraphStatistics) -> Self {
let mut mismatched_fields = Vec::new();
if incremental.computed_at_commit_epoch != recomputed.computed_at_commit_epoch {
mismatched_fields.push("computed_at_commit_epoch".to_string());
}
if incremental.node_count != recomputed.node_count {
mismatched_fields.push("node_count".to_string());
}
if incremental.relationship_count != recomputed.relationship_count {
mismatched_fields.push("relationship_count".to_string());
}
if incremental.label_counts != recomputed.label_counts {
mismatched_fields.push("label_counts".to_string());
}
if incremental.rel_type_counts != recomputed.rel_type_counts {
mismatched_fields.push("rel_type_counts".to_string());
}
Self {
ready: mismatched_fields.is_empty(),
computed_at_commit_epoch: incremental.computed_at_commit_epoch,
incremental,
recomputed,
mismatched_fields,
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct AdjacencyConsolidationPlan {
pub group_count: usize,
pub delta_entry_count: usize,
pub estimated_entries: usize,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct AdjacencyConsolidationReport {
pub planned: AdjacencyConsolidationPlan,
pub consolidated_group_count: usize,
pub consolidated_delta_entry_count: usize,
pub consolidated_estimated_entries: usize,
pub remaining: AdjacencyConsolidationPlan,
}
#[derive(Debug, Clone, Copy)]
pub struct AdjacencyConsolidationCandidate {
pub direction: AdjacencyDirection,
pub key: (NodeId, RelTypeId),
pub delta_entry_count: usize,
pub estimated_entries: usize,
}