use super::*;
pub(super) fn run_external(
fixture: &Fixture,
options: &RunOptions,
task: Option<&RuntimeTaskContext>,
) -> Outcome {
let ledger = QueryMemoryLedger::new(options.memory.query_memory_bytes);
let observer = QueryExecutionObserver::default();
let mut batches = Vec::new();
let mut live_bytes = Vec::new();
let result = GraphAlgorithmSpec {
algorithm: &options.algorithm,
graph_name: "graph",
options: &options.options,
score_column: &options.score_column,
return_node_identity: options.return_node_identity,
node_visibility_predicate: &options.predicate,
}
.stream_external(
GraphAlgorithmContext {
catalog: &fixture.catalog,
store: fixture,
memory: &options.memory,
memory_ledger: &ledger,
task_context: task,
observer: &observer,
},
ExecutionLimit {
output_rows: options.output_rows,
},
&mut |batch| {
live_bytes.push(ledger.snapshot().used_bytes);
batches.push(batch);
match options.exit {
Exit::Complete => Ok(BatchControl::Continue),
Exit::Stop => Ok(BatchControl::Stop),
Exit::Error => Err(HawDBError::StorageIntegrity("consumer sentinel".into())),
}
},
);
let memory = ledger.snapshot();
assert_eq!(memory.used_bytes, 0);
assert!(memory.classes.iter().all(|class| class.used_bytes == 0));
Outcome {
result,
batches,
live_bytes,
reports: observer.into_reports(),
peak_bytes: memory.peak_bytes,
}
}
fn dense_fixture() -> Fixture {
let mut fixture = Fixture::new();
fixture.nodes = (0..128)
.map(|id| NodeRecord {
id: NodeId(id),
labels: BTreeSet::from([LabelId(0)]),
properties: BTreeMap::new(),
})
.collect();
fixture.relationships.clear();
for source in 0..128 {
for target in 0..128 {
if source != target {
fixture.relationships.push(RelRecord {
id: RelId(fixture.relationships.len() as u64),
source: NodeId(source),
target: NodeId(target),
rel_type: RelTypeId(0),
properties: BTreeMap::new(),
});
}
}
}
fixture
}
fn predicate_fixture(prefilter: bool) -> Fixture {
use hawdb_storage::projection::ProjectedRelationshipPredicate as Filter;
let mut fixture = dense_fixture();
for node in &mut fixture.nodes {
node.properties
.insert("id".into(), Value::String(format!("memory-{}", node.id.0)));
}
for relationship in &mut fixture.relationships {
let active = !relationship.target.0.is_multiple_of(13);
let confidence =
if relationship.source.0.is_multiple_of(7) && relationship.target.0.is_multiple_of(3) {
0.1
} else {
0.9
};
relationship.properties = BTreeMap::from([
("active".into(), Value::Bool(active)),
("confidence".into(), Value::Float(confidence)),
]);
}
if prefilter {
fixture.relationships.retain(|relationship| {
!relationship.target.0.is_multiple_of(13)
&& !(relationship.source.0.is_multiple_of(7)
&& relationship.target.0.is_multiple_of(3))
});
} else {
fixture.definition.as_mut().unwrap().relationship_predicates = BTreeMap::from([(
"LINK".into(),
Filter::And(vec![
Filter::Eq {
property: "active".into(),
value: Value::Bool(true),
},
Filter::Gte {
property: "confidence".into(),
value: Value::Float(0.5),
},
]),
)]);
}
fixture
}
#[test]
fn resident_and_streaming_preserve_predicates_options_and_complete_identities() {
let fixture = predicate_fixture(false);
let control = predicate_fixture(true);
assert!(control.relationships.len() < fixture.relationships.len());
for algorithm in [GraphAlgorithmKind::PageRank, GraphAlgorithmKind::Louvain] {
let mut options = RunOptions {
algorithm,
return_node_identity: true,
..RunOptions::default()
};
options.options.tolerance = Some(0.2);
options.options.normalize_initial = Some(false);
options.options.resolution = Some(0.5);
options.memory.batch_rows = nz(7);
options.memory.blocking_operator_bytes = nz(16 * 1024 * 1024);
options.memory.query_memory_bytes = nz(32 * 1024 * 1024);
let resident = run(&fixture, &options, None);
assert_eq!(resident.result.unwrap(), BatchControl::Continue);
assert_eq!(
resident.reports.blocking_memory[0].operator,
"GraphAlgorithm"
);
let expected = run(&control, &options, None);
assert_eq!(expected.result.unwrap(), BatchControl::Continue);
let resident: Vec<_> = resident.batches.into_iter().flatten().collect();
let expected: Vec<_> = expected.batches.into_iter().flatten().collect();
assert_eq!(resident, expected);
options.memory.blocking_operator_bytes = nz(96 * 1024);
let external = run(&fixture, &options, None);
assert_eq!(external.result.unwrap(), BatchControl::Continue);
assert_eq!(
external.reports.blocking_memory[0].operator,
"GraphAlgorithmStreaming"
);
assert!(external.reports.blocking_memory[0].peak_tracked_bytes <= 96 * 1024);
let external: Vec<_> = external.batches.into_iter().flatten().collect();
assert_eq!(external, resident);
assert_eq!(
external.len(),
128,
"stationary Louvain phases must not duplicate assignments"
);
for row in external {
let Value::Int(node) = row.values["node"] else {
panic!("missing node")
};
assert_eq!(
row.values["node_id"],
Value::String(format!("memory-{node}"))
);
assert_eq!(row.values["node_label"], Value::String("Memory".into()));
}
}
}
#[test]
fn late_streaming_identity_failure_emits_no_prefix() {
let mut fixture = predicate_fixture(false);
fixture.fail_identity_node = Some(NodeId(127));
for algorithm in [GraphAlgorithmKind::PageRank, GraphAlgorithmKind::Louvain] {
let mut options = RunOptions {
algorithm,
return_node_identity: true,
..RunOptions::default()
};
options.memory.blocking_operator_bytes = nz(96 * 1024);
let output = run(&fixture, &options, None);
assert!(output
.result
.unwrap_err()
.to_string()
.contains("identity lookup sentinel"));
assert!(output.batches.is_empty());
assert_eq!(
output.reports.blocking_memory[0].operator,
"GraphAlgorithmStreaming"
);
}
}
#[test]
fn both_algorithms_complete_when_edges_exceed_the_blocking_budget() {
let fixture = dense_fixture();
let budget = 96 * 1024;
assert!(fixture.relationships.len() * std::mem::size_of::<usize>() > budget);
for algorithm in [GraphAlgorithmKind::PageRank, GraphAlgorithmKind::Louvain] {
let mut options = RunOptions {
algorithm,
..RunOptions::default()
};
options.options.max_iterations = Some(3);
options.options.max_levels = Some(2);
options.memory.blocking_operator_bytes = nz(budget);
options.memory.batch_rows = nz(7);
let layout = if algorithm == GraphAlgorithmKind::PageRank {
ProjectionLayout::Outgoing
} else {
ProjectionLayout::Undirected
};
assert!(try_projected_graph_with_node_filter(
&fixture.catalog,
&fixture,
&[],
&[],
|_| true,
layout,
ProjectionMemoryBudget::new(nz(budget))
)
.is_err());
let output = run(&fixture, &options, None);
assert_eq!(output.result.unwrap(), BatchControl::Continue);
let rows: Vec<_> = output.batches.into_iter().flatten().collect();
assert_eq!(
rows.len(),
128,
"stationary Louvain phases must not duplicate assignments"
);
for row in rows {
if algorithm == GraphAlgorithmKind::PageRank {
let Value::Float(score) = row.values[&options.score_column] else {
panic!("missing score")
};
assert!((score - 1.0 / 128.0).abs() < 1e-14);
} else {
assert_eq!(row.values["louvain_id"], Value::Int(0));
}
}
assert!(fixture.adjacency_visits.get() > fixture.relationships.len());
let report = &output.reports.blocking_memory[0];
assert_eq!(report.operator, "GraphAlgorithmStreaming");
assert!(report.peak_tracked_bytes <= budget);
assert!(output.peak_bytes <= budget);
assert_eq!(report.spilled_bytes, 0);
}
}
#[test]
fn cancellation_and_storage_failure_emit_no_partial_algorithm_results() {
for algorithm in [GraphAlgorithmKind::PageRank, GraphAlgorithmKind::Louvain] {
for fail_at in [0, 127, 2048] {
let mut fixture = dense_fixture();
let mut options = RunOptions {
algorithm,
..RunOptions::default()
};
options.memory.blocking_operator_bytes = nz(96 * 1024);
fixture.fail_adjacency_at = Some(fail_at);
let output = run_external(&fixture, &options, None);
assert!(output
.result
.unwrap_err()
.to_string()
.contains("adjacency scan sentinel"));
assert!(output.batches.is_empty());
let token = RuntimeCancellationToken::new();
let task = RuntimeTaskContext::without_deadline(token.clone());
fixture.fail_adjacency_at = None;
fixture.adjacency_visits.set(0);
fixture.cancel_adjacency_at = Some((fail_at, token));
let output = run_external(&fixture, &options, Some(&task));
assert!(output
.result
.unwrap_err()
.to_string()
.contains("runtime task stopped: cancelled"));
assert!(output.batches.is_empty());
}
}
}
#[test]
fn node_state_and_oversized_records_remain_fail_closed() {
let mut fixture = Fixture::new();
let mut options = RunOptions::default();
options.memory.blocking_operator_bytes = nz(512);
let output = run_external(&fixture, &options, None);
assert!(output
.result
.unwrap_err()
.to_string()
.contains("streaming node scan"));
assert!(output.batches.is_empty());
options.memory.blocking_operator_bytes = nz(4096);
fixture.relationships[0]
.properties
.insert("large".into(), Value::String("x".repeat(8192)));
let output = run_external(&fixture, &options, None);
assert!(output
.result
.unwrap_err()
.to_string()
.contains("adjacency record"));
assert!(output.batches.is_empty());
}