crates/knut-popviz/src/lib.rs
//! Population bulk-loader **driver + introspectable state** — the headless core
//! of the egui `knut-popviz` app. Generates the synthetic population as Arrow,
//! bulk-loads it through `graphar-falkordb`'s `FalkorDbLoader`, and exposes its
//! progress as a [`PopvizState`] whose [`PopvizState::state_json`] is the same
//! external-state contract the pipeline viewer uses — so the GUI can be
//! robot-tested headlessly (assert what's on screen without a display).
//!
//! The egui app (`src/main.rs`, feature `gui`) is a thin shell over this.
/// The short git commit this crate was built from (`KNUT_GIT_HASH`, emitted by
/// `build.rs` via a pure-fs, zero-shell read of the repo's git metadata). `env!`
/// (not `option_env!`) because `build.rs` emits it on every build.
pub const GIT_HASH: &str = env!("KNUT_GIT_HASH");
/// `--version` line: `v<pkg-version> (<git-sha>)`. The `knut-popviz` bin prints
/// `knut-popviz <this>` for `--version` — the suite-wide identity.
pub const VERSION_LINE: &str = concat!(
"v",
env!("CARGO_PKG_VERSION"),
" (",
env!("KNUT_GIT_HASH"),
")"
);
use std::sync::{Arc, Mutex};
use std::time::Instant;
use arrow_array::{BooleanArray, Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use graphar_falkordb::loader::choose_bulk;
use graphar_falkordb::{BulkEdgeGroup, BulkNodeGroup, FalkorDbLoader};
// The pure population model — labels, counts, geo placement, the canonical
// generic graph — lives in `knut-popsim` (wasm-clean, shared with the facett
// demo). popviz re-exports it and adds the Arrow → FalkorDB/Iceberg load path.
pub use knut_popsim::geo;
// The desktop UI itself (feature `gui`). It lives in the LIBRARY, not in
// `src/main.rs`, so an integration test can link it: the binary is unreachable
// from `tests/`, which is why nothing in this crate had ever asserted a click
// or a rendered pixel. `main.rs` is the thin `run_native` shell over this.
#[cfg(feature = "gui")]
pub mod app;
#[cfg(feature = "iceberg")]
pub mod permutate;
pub mod lake;
/// **Introspection / emit marker** — record one functional-status row for the
/// nornir test matrix. Wraps `nornir_testmatrix::functional_status` behind the
/// `testmatrix` feature (a compiled-out `#[inline]` no-op otherwise, with no
/// nornir dep). `run_load` calls this so `nornir test --features testmatrix`
/// SEES the bulk-loader's health.
#[inline]
pub fn functional_status(component: &str, check: &str, ok: bool, detail: &str) {
#[cfg(feature = "testmatrix")]
nornir_testmatrix::functional_status(component, check, ok, detail);
#[cfg(not(feature = "testmatrix"))]
{
let _ = (component, check, ok, detail);
}
}
pub use knut_popsim::{EKONOMISKT_BISTAND_RATE, LABELS, PERSON_REL};
/// Labels that carry a Swedish address (lat/lon/kommun). `Person` additionally
/// carries the `ekonomiskt_bistand` flag the cluster pass keys on.
fn label_has_geo(label: &str) -> bool {
matches!(label, "Person" | "Address" | "Property")
}
const CHUNK: i64 = 25_000;
/// Per-label load progress.
#[derive(Debug, Clone)]
pub struct LabelProgress {
pub name: String,
pub loaded: usize,
pub total: usize,
}
/// What phase the driver is in.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Phase {
Idle,
Vertices,
Edges,
Indexes,
Done,
Failed,
}
impl Phase {
pub fn label(self) -> &'static str {
match self {
Phase::Idle => "idle",
Phase::Vertices => "loading vertices",
Phase::Edges => "loading edges",
Phase::Indexes => "indexing",
Phase::Done => "done",
Phase::Failed => "failed",
}
}
/// Parse a phase back from its [`label`](Self::label) string — the command
/// seam's `set_phase` atom. Unknown ⇒ `None`.
pub fn from_label(s: &str) -> Option<Phase> {
Some(match s {
"idle" => Phase::Idle,
"loading vertices" => Phase::Vertices,
"loading edges" => Phase::Edges,
"indexing" => Phase::Indexes,
"done" => Phase::Done,
"failed" => Phase::Failed,
_ => return None,
})
}
}
/// The canonical FalkorDB load order: **vertices → indexes → edges**.
///
/// The `_gar_id` indexes are built BEFORE the edge phase on purpose. Every edge
/// query `MATCH (a:`Label` {`_gar_id`: …})`es its two endpoints (see
/// `graphar_falkordb::cypher::unwind_edge_query_keyed`); with no index on
/// `_gar_id` each MATCH is a full O(V) label scan, so the whole edge phase is
/// O(edges × vertices) work *on the server* — the one stall the client-side
/// pipelining (`insert_*_par`) cannot touch, because it is the server doing a
/// quadratic scan, not the client waiting on round-trips. Building the index
/// first turns every endpoint MATCH into an indexed O(1) lookup, so the edge
/// phase is O(edges). Vertices come first (nothing to index until the nodes
/// exist); index maintenance never touches the edge CREATE (edges are
/// relationships, `_gar_id` is a node property), so there is no downside to
/// indexing before — it is a pure win.
///
/// [`load_falkor`] drives its whole phase sequence off this array, so this
/// ordering is the single enforced contract (asserted DB-free in
/// `falkor_indexes_are_built_before_edges`).
pub fn falkor_load_phase_order() -> [Phase; 3] {
[Phase::Vertices, Phase::Indexes, Phase::Edges]
}
/// Per-relationship edge-load progress. Every non-`Person` label carries exactly
/// one `Person -> <label>` relationship (parallel to [`LABELS`] / [`PERSON_REL`]),
/// so `rels[k]` describes the edge into `LABELS[k + 1]`.
#[derive(Debug, Clone)]
pub struct RelProgress {
/// The relationship type (a value from [`PERSON_REL`], e.g. `LIVES_AT`).
pub name: String,
/// Destination label name (`Person` is always the source).
pub dst_label: String,
/// Index into [`LABELS`] of the destination label.
pub dst_li: usize,
pub loaded: usize,
pub total: usize,
}
/// The whole observable state of a population load — shared (`Arc<Mutex<…>>`)
/// between the driver thread and the UI, and dumped as JSON for robot tests.
#[derive(Debug, Clone)]
pub struct PopvizState {
pub phase: Phase,
pub labels: Vec<LabelProgress>,
/// Per-relationship edge progress — one `Person -> <label>` rel per non-Person
/// label. Populated during [`Phase::Edges`].
pub rels: Vec<RelProgress>,
pub elapsed_s: f64,
pub error: Option<String>,
/// The sink the load is going to (`falkor` / `iceberg`).
pub target: String,
}
impl PopvizState {
/// A fresh state sized for `scale` (per-label totals computed, nothing loaded).
pub fn new(scale: f64) -> Self {
let labels: Vec<LabelProgress> = LABELS
.iter()
.map(|(name, n)| LabelProgress {
name: (*name).to_string(),
loaded: 0,
total: ((*n as f64) * scale).max(1.0) as usize,
})
.collect();
// One `Person -> <label>` relationship per non-Person label (li 1..). Its
// edge total is the destination label's vertex count: exactly one edge per
// entity (each entity is owned by a person — see [`edge_batches`]).
let rels: Vec<RelProgress> = (1..LABELS.len())
.map(|li| RelProgress {
name: PERSON_REL[li].to_string(),
dst_label: LABELS[li].0.to_string(),
dst_li: li,
loaded: 0,
total: labels[li].total,
})
.collect();
Self {
phase: Phase::Idle,
labels,
rels,
elapsed_s: 0.0,
error: None,
target: "falkor".into(),
}
}
/// Total edges to load across every relationship.
pub fn edges_total(&self) -> usize {
self.rels.iter().map(|r| r.total).sum()
}
/// Edges loaded so far across every relationship.
pub fn edges_loaded(&self) -> usize {
self.rels.iter().map(|r| r.loaded).sum()
}
pub fn total(&self) -> usize {
self.labels.iter().map(|l| l.total).sum()
}
pub fn loaded(&self) -> usize {
self.labels.iter().map(|l| l.loaded).sum()
}
pub fn fraction(&self) -> f32 {
let t = self.total();
if t == 0 {
0.0
} else {
self.loaded() as f32 / t as f32
}
}
pub fn rows_per_s(&self) -> f64 {
if self.elapsed_s > 0.0 {
self.loaded() as f64 / self.elapsed_s
} else {
0.0
}
}
/// The introspection snapshot — assert on this headlessly (no display).
/// **Rule:** any new visible field must be added here.
pub fn state_json(&self) -> serde_json::Value {
let per_label: serde_json::Map<String, serde_json::Value> = self
.labels
.iter()
.map(|l| {
(
l.name.clone(),
serde_json::json!({ "loaded": l.loaded, "total": l.total }),
)
})
.collect();
let per_rel: serde_json::Map<String, serde_json::Value> = self
.rels
.iter()
.map(|r| {
(
r.name.clone(),
serde_json::json!({
"loaded": r.loaded,
"total": r.total,
"src": "Person",
"dst": r.dst_label,
}),
)
})
.collect();
let mut placeholders = Vec::new();
if self.phase == Phase::Idle {
placeholders.push("not started");
}
serde_json::json!({
"phase": self.phase.label(),
"target": self.target,
"total": self.total(),
"loaded": self.loaded(),
"fraction": self.fraction(),
"rows_per_s": self.rows_per_s(),
"elapsed_s": self.elapsed_s,
"labels": per_label,
"edges_total": self.edges_total(),
"edges_loaded": self.edges_loaded(),
"edges": per_rel,
"error": self.error,
"placeholders": placeholders,
})
}
/// **Command seam** (LAW: every UI atom must be remote-controllable as data).
/// Drives the observable popviz atoms — **select target · set phase · advance
/// the population view · reset** — purely in-memory (no load, no network), so a
/// robot scenario drives the UI headlessly and a visible `watch` run renders
/// the same shared state. Unknown or garbled input is ignored — never panics.
pub fn update_json(&mut self, msg: &str) {
let Ok(v) = serde_json::from_str::<serde_json::Value>(msg) else {
return;
};
if let Some(t) = v.get("select_target").and_then(|t| t.as_str()) {
if t == "falkor" || t == "iceberg" {
self.target = t.to_string();
}
}
if let Some(p) = v.get("set_phase").and_then(|p| p.as_str()) {
if let Some(ph) = Phase::from_label(p) {
self.phase = ph;
}
}
if let Some(adv) = v.get("advance").and_then(|a| a.as_object()) {
if let (Some(name), Some(n)) = (
adv.get("label").and_then(|x| x.as_str()),
adv.get("loaded").and_then(|x| x.as_u64()),
) {
if let Some(l) = self.labels.iter_mut().find(|l| l.name == name) {
l.loaded = (n as usize).min(l.total);
}
}
}
if v.get("reset").and_then(|r| r.as_bool()) == Some(true) {
self.phase = Phase::Idle;
self.error = None;
self.elapsed_s = 0.0;
for l in &mut self.labels {
l.loaded = 0;
}
for r in &mut self.rels {
r.loaded = 0;
}
}
}
}
/// Where the simulated population lands.
#[derive(Debug, Clone)]
pub enum Target {
/// FalkorDB graph (via `graphar-falkordb`'s `FalkorDbLoader`).
Falkor,
/// An Iceberg warehouse (one table per label, via `skade`) — needs the
/// `iceberg` feature.
Iceberg { warehouse: String },
/// The same tables behind an Iceberg REST catalog (MinIO AIStor `/_iceberg`) - needs
/// the `iceberg-rest` feature. See [`lake`].
IcebergRest(lake::RestLake),
}
impl Target {
pub fn label(&self) -> &'static str {
match self {
Target::Falkor => "falkor",
Target::Iceberg { .. } => "iceberg",
Target::IcebergRest(_) => "iceberg-rest",
}
}
}
/// Which FalkorDB ingest lane [`load_falkor`] takes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum IngestMode {
/// Native `GRAPH.BULK` for an empty/fresh graph, parallel-UNWIND otherwise —
/// the production default (`graphar_falkordb::loader::choose_bulk` on
/// `graph_is_empty`).
#[default]
Auto,
/// Always the native `GRAPH.BULK` fast lane (fails if the graph is non-empty,
/// since bulk `CREATE`s and cannot MERGE). For the parity test's bulk arm.
Bulk,
/// Always the parallel-UNWIND path — the proven incremental lane, and the
/// parity test's reference arm.
Unwind,
}
impl IngestMode {
/// Parse the `--ingest` CLI value; unknown ⇒ [`IngestMode::Auto`].
pub fn from_str_lenient(s: &str) -> IngestMode {
match s {
"bulk" => IngestMode::Bulk,
"unwind" => IngestMode::Unwind,
_ => IngestMode::Auto,
}
}
}
/// Connection + scale parameters for a load.
#[derive(Debug, Clone)]
pub struct LoadParams {
pub url: String,
pub graph: String,
pub scale: f64,
pub drop: bool,
/// Sink: FalkorDB graph or an Iceberg warehouse.
pub target: Target,
/// FalkorDB ingest lane selection (bulk fast-lane vs UNWIND). Ignored by the
/// Iceberg target.
pub ingest: IngestMode,
/// A mutual-TLS client identity for a `rediss://` FalkorDB (`None` = plain `redis://`).
/// A server running `tls-auth-clients yes` refuses a load without one, and with
/// `tls-auth-clients-user CN` the certificate's CN is the user the load runs as.
/// Ignored by the Iceberg target.
pub tls: Option<graphar_falkordb::FalkorTls>,
}
/// One label's vertices as Arrow batches of ≤`CHUNK` rows. Always `_gar_id` +
/// `idx` + `name`; **geo-bearing labels** (`Person`/`Address`/`Property`, see
/// [`label_has_geo`]) additionally carry a realistic Swedish address —
/// `lon`/`lat` (f64) + `kommun` (utf8) — and `Person` carries the
/// `ekonomiskt_bistand` (benefit-recipient) flag the cluster pass keys on. Those
/// columns flow verbatim through skade→Iceberg and FalkorDB.
///
/// The **serial** per-label generator — production now fans row-chunks across all
/// cores via [`gen_all_vertex_batches`]; this remains as the serial reference the
/// byte-identity test checks that fan-out against (both call [`vertex_batch_range`]).
#[cfg(test)]
fn vertex_batches(label: &str, count: i64, offset: i64) -> Vec<RecordBatch> {
let schema = vertex_schema(label);
let mut out = Vec::new();
let mut start = 0i64;
while start < count {
let end = (start + CHUNK).min(count);
out.push(vertex_batch_range(label, &schema, start, end, offset));
start = end;
}
out
}
/// The Arrow schema for a label's vertex batches (shared by every chunk of the
/// same label). Factored out of [`vertex_batches`] so the parallel row-chunk
/// generator can build it once per label and hand the same `schema` to every
/// chunk — identical to the serial path's per-label schema.
fn vertex_schema(label: &str) -> Arc<Schema> {
let geo = label_has_geo(label);
let mut fields = vec![
Field::new("_gar_id", DataType::Int64, false),
Field::new("idx", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
];
if geo {
fields.push(Field::new("lon", DataType::Float64, false));
fields.push(Field::new("lat", DataType::Float64, false));
fields.push(Field::new("kommun", DataType::Utf8, false));
}
if label == "Person" {
fields.push(Field::new("ekonomiskt_bistand", DataType::Boolean, false));
}
Arc::new(Schema::new(fields))
}
/// Generate **one** ≤`CHUNK`-row vertex batch for rows `[start, end)` of `label`,
/// with each row's `_gar_id` = `offset + i` and `idx`/`name` = `i` (byte-identical
/// to what [`vertex_batches`]' serial loop produced for the same `[start, end)`
/// slice). This is the per-row-range work unit the gatling fan-out dispatches, so
/// a single dominant label's rows split into many balanced chunks across cores.
fn vertex_batch_range(
label: &str,
schema: &Arc<Schema>,
start: i64,
end: i64,
offset: i64,
) -> RecordBatch {
let geo = label_has_geo(label);
let ids: Vec<i64> = (start..end).map(|i| offset + i).collect();
let idx: Vec<i64> = (start..end).collect();
let names: Vec<String> = (start..end).map(|i| format!("n{i}")).collect();
let mut cols: Vec<arrow_array::ArrayRef> = vec![
Arc::new(Int64Array::from(ids.clone())),
Arc::new(Int64Array::from(idx)),
Arc::new(StringArray::from(names)),
];
if geo {
let addrs: Vec<geo::Address> = ids.iter().map(|&id| geo::address_for(id)).collect();
cols.push(Arc::new(Float64Array::from(
addrs.iter().map(|a| a.lon).collect::<Vec<_>>(),
)));
cols.push(Arc::new(Float64Array::from(
addrs.iter().map(|a| a.lat).collect::<Vec<_>>(),
)));
cols.push(Arc::new(StringArray::from(
addrs
.iter()
.map(|a| a.kommun.to_string())
.collect::<Vec<_>>(),
)));
}
if label == "Person" {
cols.push(Arc::new(BooleanArray::from(
ids.iter()
.map(|&id| geo::is_ekonomiskt_bistand(id, EKONOMISKT_BISTAND_RATE))
.collect::<Vec<_>>(),
)));
}
RecordBatch::try_new(schema.clone(), cols).expect("batch")
}
/// The flat vertex work-plan: one balanced ≤`CHUNK` row-range unit per
/// `(li, [start, end))`, in `(label ascending, row ascending)` order — the
/// decompressor idiom (many balanced units, one per core) so a single dominant
/// label saturates all cores instead of stalling one (the ~35%-CPU stall where
/// the 60M-row `TaxRecord` was ONE single-threaded work unit). This flat order is
/// the single source of truth both the buffered collector
/// ([`gen_all_vertex_batches`]) and the bounded streamer ([`BatchStream`]) drive
/// off, so streamed and buffered generation produce the SAME batches in the SAME
/// order.
fn vertex_plan(totals: &[i64]) -> Vec<(usize, i64, i64)> {
let mut plan: Vec<(usize, i64, i64)> = Vec::new();
for (li, &count) in totals.iter().enumerate() {
let mut start = 0i64;
while start < count {
let end = (start + CHUNK).min(count);
plan.push((li, start, end));
start = end;
}
}
plan
}
/// The flat edge work-plan — the [`vertex_plan`] sibling for the one
/// `Person -> <label>` relationship per non-`Person` label (`li` in `1..`, since
/// `Person` has no rel). Same balanced ≤`CHUNK` shape and `(label, row)` order.
fn edge_plan(totals: &[i64]) -> Vec<(usize, i64, i64)> {
let mut plan: Vec<(usize, i64, i64)> = Vec::new();
for li in 1..LABELS.len() {
let mut start = 0i64;
while start < totals[li] {
let end = (start + CHUNK).min(totals[li]);
plan.push((li, start, end));
start = end;
}
}
plan
}
/// Window size for the bounded streaming generators: one gatling fan-out worth of
/// batches (≈ one balanced row-range per core) generated + held at a time. Keeps
/// every core busy per window (LAW #0: gatling) while capping peak memory to a
/// single window of Arrow batches instead of the whole population.
fn stream_window() -> usize {
znippy_zoomies::gatling_forkjoin::default_workers().max(1)
}
/// A **bounded-memory, gatling-parallel** batch stream. Yields the batches for a
/// flat work-plan (see [`vertex_plan`]/[`edge_plan`]) IN PLAN ORDER, materializing
/// at most one `window` of units at a time: each window is fanned across all cores
/// via `gatling_for_each` (LAW #0: gatling, never rayon), buffered, handed to the
/// consumer one batch at a time, and dropped before the next window is generated.
///
/// Peak memory is therefore **one window** of Arrow batches, NOT the whole graph —
/// the fix for the korp OOM, where the loader now pulls batches incrementally
/// instead of the generator collecting the entire population first. Draining the
/// stream to exhaustion yields the EXACT flat sequence a monolithic
/// `gatling_for_each(plan.len(), …)` would produce (each unit runs the same
/// `make(li, start, end)`, windows consumed in plan order), so the generated graph
/// is byte-identical to the buffered path (proven RED-when-broken in
/// `stream_matches_buffered_byte_identical`).
struct BatchStream<'a, F>
where
F: Fn(usize, i64, i64) -> RecordBatch + Sync,
{
plan: &'a [(usize, i64, i64)],
make: F,
window: usize,
cursor: usize,
buf: std::collections::VecDeque<(usize, RecordBatch)>,
}
impl<'a, F> BatchStream<'a, F>
where
F: Fn(usize, i64, i64) -> RecordBatch + Sync,
{
fn new(plan: &'a [(usize, i64, i64)], make: F, window: usize) -> Self {
Self {
plan,
make,
window: window.max(1),
cursor: 0,
buf: std::collections::VecDeque::new(),
}
}
}
impl<F> Iterator for BatchStream<'_, F>
where
F: Fn(usize, i64, i64) -> RecordBatch + Sync,
{
/// `(destination label index, batch)` — `li` matches the plan's label index.
type Item = (usize, RecordBatch);
fn next(&mut self) -> Option<Self::Item> {
if self.buf.is_empty() && self.cursor < self.plan.len() {
use znippy_zoomies::gatling_forkjoin::gatling_for_each;
let start = self.cursor;
let end = (start + self.window).min(self.plan.len());
// Copy the ≤`window` units out so the gatling fan-out (and the later
// `self.buf`/`self.cursor` mutation) don't hold a borrow of `self`.
let units: Vec<(usize, i64, i64)> = self.plan[start..end].to_vec();
let make = &self.make;
// One core-wide gatling fan per window (LAW #0: gatling, never rayon).
let batches: Vec<RecordBatch> = gatling_for_each(units.len(), 0, |k| {
let (li, s, e) = units[k];
make(li, s, e)
});
for ((li, _, _), b) in units.into_iter().zip(batches) {
self.buf.push_back((li, b));
}
self.cursor = end;
}
self.buf.pop_front()
}
}
/// Collect every label's vertex batches into per-label vecs (result index `[li]`).
/// This is the **buffered** collector — it drains a [`BatchStream`] to exhaustion,
/// so it produces the identical batches the streaming load paths do, just held all
/// at once. Kept for the `GRAPH.BULK` fast lane ([`load_falkor_bulk`]), whose
/// encoder needs every blob's batches materialized as slices simultaneously; the
/// UNWIND and Iceberg lanes stream instead (bounded memory). (LAW #0: gatling.)
fn gen_all_vertex_batches(totals: &[i64], offsets: &[i64]) -> Vec<Vec<RecordBatch>> {
// Per-label schema, built once (a chunk of a label reuses its label's schema).
let schemas: Vec<Arc<Schema>> = LABELS.iter().map(|(l, _)| vertex_schema(l)).collect();
let plan = vertex_plan(totals);
let make = |li: usize, start: i64, end: i64| {
vertex_batch_range(LABELS[li].0, &schemas[li], start, end, offsets[li])
};
let mut out: Vec<Vec<RecordBatch>> = (0..LABELS.len()).map(|_| Vec::new()).collect();
for (li, b) in BatchStream::new(&plan, make, stream_window()) {
out[li].push(b);
}
out
}
/// Buffered edge sibling of [`gen_all_vertex_batches`] — drains an edge
/// [`BatchStream`] into per-relationship vecs, indexed by `li - 1` (labels `1..`,
/// since `Person` has no rel). `person_offset` and the per-label `entity_offset`
/// are passed through verbatim, so the produced edges are byte-identical to the
/// serial [`edge_batches`] output (the bulk lane passes zero offsets, the
/// UNWIND/Iceberg lanes pass the real cumulative offsets). Kept for the bulk lane;
/// UNWIND/Iceberg stream the same edge plan instead.
fn gen_all_edge_batches(
totals: &[i64],
person_offset: i64,
entity_offset: &[i64],
) -> Vec<Vec<RecordBatch>> {
let plan = edge_plan(totals);
let person_count = totals[0];
let make = |li: usize, start: i64, end: i64| {
edge_batch_range(person_count, person_offset, entity_offset[li], start, end)
};
let mut out: Vec<Vec<RecordBatch>> = (0..LABELS.len() - 1).map(|_| Vec::new()).collect();
for (li, b) in BatchStream::new(&plan, make, stream_window()) {
out[li - 1].push(b);
}
out
}
/// The `Person -> <entity>` edges for destination label index `li` (never
/// `PERSON`), as Arrow batches of ≤`CHUNK` rows carrying **only** `src`/`dst`
/// Int64 columns — GraphAr `_gar_id` values, the keys `insert_edges_keyed`
/// matches endpoints on. One edge per entity: entity `j` is owned by person
/// `j % person_count`, so every entity is connected and the layout is fully
/// deterministic (robot tests can assert exact edge counts). The relationship
/// name is `PERSON_REL[li]` and is passed as the edge label, not a column.
///
/// The **serial** per-relationship generator — production fans row-chunks across
/// cores via [`gen_all_edge_batches`]; kept as the serial byte-identity reference.
#[cfg(test)]
fn edge_batches(
person_count: i64,
person_offset: i64,
entity_count: i64,
entity_offset: i64,
) -> Vec<RecordBatch> {
let mut out = Vec::new();
let mut start = 0i64;
while start < entity_count {
let end = (start + CHUNK).min(entity_count);
out.push(edge_batch_range(
person_count,
person_offset,
entity_offset,
start,
end,
));
start = end;
}
out
}
/// The Arrow schema for edge batches — just the `src`/`dst` `_gar_id` endpoint
/// key columns (see [`edge_batches`]).
fn edge_schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
]))
}
/// Generate **one** ≤`CHUNK`-row edge batch for entity rows `[start, end)`:
/// `dst = entity_offset + j`, `src = person_offset + (j % person_count)` — the
/// per-row-range work unit the gatling edge fan-out dispatches, byte-identical to
/// the slice [`edge_batches`]' serial loop produced for the same `[start, end)`.
fn edge_batch_range(
person_count: i64,
person_offset: i64,
entity_offset: i64,
start: i64,
end: i64,
) -> RecordBatch {
let persons = person_count.max(1);
let src: Vec<i64> = (start..end)
.map(|j| person_offset + (j % persons))
.collect();
let dst: Vec<i64> = (start..end).map(|j| entity_offset + j).collect();
RecordBatch::try_new(
edge_schema(),
vec![
Arc::new(Int64Array::from(src)),
Arc::new(Int64Array::from(dst)),
],
)
.expect("edge batch")
}
/// Run the bulk load, updating `state` and calling `wake` after each batch (so
/// the UI repaints). Blocking — the caller runs it on its own thread. On the
/// happy path it ends in [`Phase::Done`]; any error lands in [`Phase::Failed`]
/// with `error` set.
pub fn run_load(
params: LoadParams,
state: Arc<Mutex<PopvizState>>,
wake: impl Fn() + Send + 'static,
) {
let rt = match tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
{
Ok(rt) => rt,
Err(e) => {
fail(&state, format!("runtime: {e}"));
wake();
return;
}
};
state.lock().unwrap().target = params.target.label().to_string();
let t0 = Instant::now();
rt.block_on(async move {
let totals: Vec<i64> = {
state
.lock()
.unwrap()
.labels
.iter()
.map(|l| l.total as i64)
.collect()
};
let mut offsets = vec![0i64; totals.len()];
let mut acc = 0i64;
for (i, c) in totals.iter().enumerate() {
offsets[i] = acc;
acc += c;
}
set_phase(&state, Phase::Vertices);
let result = match ¶ms.target {
Target::Falkor => load_falkor(¶ms, &state, &totals, &offsets, &t0, &wake).await,
Target::Iceberg { warehouse } => {
load_iceberg(
&lake::Lake::Dir(warehouse.into()),
params.drop,
&state,
&totals,
&offsets,
&t0,
&wake,
)
.await
}
Target::IcebergRest(r) => {
load_iceberg(&lake::Lake::Rest(r.clone()), params.drop, &state, &totals, &offsets, &t0, &wake).await
}
};
match result {
Ok(()) => {
let mut s = state.lock().unwrap();
s.elapsed_s = t0.elapsed().as_secs_f64();
s.phase = Phase::Done;
drop(s);
functional_status(
"knut-popviz/run_load",
"population_load",
true,
params.target.label(),
);
}
Err(msg) => {
functional_status("knut-popviz/run_load", "population_load", false, &msg);
fail(&state, msg);
}
}
wake();
});
}
/// Load the whole synthetic population into FalkorDB, auto-selecting the ingest
/// lane. `--drop`'s `DETACH DELETE` runs FIRST (a barrier), then:
///
/// * a **fresh/empty** graph takes the native `GRAPH.BULK` fast lane
/// ([`load_falkor_bulk`]) — benched ~48× the parallel-UNWIND path; or
/// * a **non-empty** graph keeps the proven incremental parallel-UNWIND path
/// ([`load_falkor_unwind`]) — bulk `CREATE`s and cannot MERGE onto live nodes.
///
/// The decision is `graphar_falkordb::loader::choose_bulk(graph_is_empty)` in
/// [`IngestMode::Auto`]; the parity test pins [`IngestMode::Bulk`] /
/// [`IngestMode::Unwind`] to load the SAME population both ways and assert the
/// two graphs are byte-identical. Both lanes produce the identical graph — the
/// `_gar_id` property is the global, offset-based id either way (the UNWIND path
/// carries it in the batch column and MATCHes on it; the bulk encoder detects
/// the same column and stores it verbatim, then addresses edge endpoints by
/// group position whose base offset equals that global `_gar_id`).
async fn load_falkor(
params: &LoadParams,
state: &Arc<Mutex<PopvizState>>,
totals: &[i64],
offsets: &[i64],
t0: &Instant,
wake: &(impl Fn() + Send),
) -> Result<(), String> {
let mut loader = FalkorDbLoader::connect_with_identity(¶ms.url, ¶ms.graph, params.tls.clone())
.await
.map_err(|e| format!("connect: {e}"))?;
if params.drop {
// A failed wipe must abort — proceeding onto a dirty graph double-counts
// (vertices use CREATE, not MERGE), silently inflating the result.
// A *successful* wipe is destructive too, so announce it: `--drop`
// must never destroy an existing graph without a word to the user.
// This BARRIER stays before any insert.
eprintln!(
"knut-popviz: --drop wiping existing graph {:?} before load",
params.graph
);
loader
.run_query("MATCH (n) DETACH DELETE n")
.await
.map_err(|e| format!("drop graph: {e}"))?;
}
let use_bulk = match params.ingest {
IngestMode::Bulk => true,
IngestMode::Unwind => false,
// The production auto-select: an empty graph is a fresh initial
// population → bulk fast lane; anything with nodes → incremental UNWIND.
IngestMode::Auto => choose_bulk(loader.graph_is_empty().await),
};
if use_bulk {
load_falkor_bulk(&mut loader, state, totals, offsets, t0, wake).await
} else {
load_falkor_unwind(&mut loader, state, totals, offsets, t0, wake).await
}
}
/// **Spread a cumulative "entities loaded so far" count over per-group totals, in
/// order** — the mapping the streamed `GRAPH.BULK` lane uses to turn its single
/// cumulative counter into live per-label / per-relationship progress.
///
/// Groups are sent to the server in `LABELS` order and each is its own blob, so the
/// first `done` entities are exactly the first groups filled to their totals, with at
/// most one partially-filled group at the frontier. That makes this an EXACT
/// reconstruction of what has landed, not an interpolation — which matters, because a
/// progress bar driven by a guess is worse than one that jumps.
///
/// Pure ⇒ unit-testable with no server.
pub fn distribute_progress(totals: &[usize], done: usize) -> Vec<usize> {
let mut left = done;
totals
.iter()
.map(|&t| {
let take = left.min(t);
left -= take;
take
})
.collect()
}
/// Native `GRAPH.BULK` initial-population fast lane — the empty-graph path
/// [`load_falkor`] auto-selects. Encodes the whole population from the SAME
/// Arrow batches the UNWIND path uses and ships it via
/// [`FalkorDbLoader::bulk_insert_streamed`] (one appendable `GRAPH.BULK` call per
/// blob — bounded per-command memory).
///
/// **Reconciling popviz's `_gar_id` scheme with the loader-level bulk API:**
/// * *Nodes* — popviz's [`vertex_batches`] already carry a `_gar_id` column
/// holding the **global, offset-based** id (`offsets[li] + row`). The bulk
/// encoder detects that column and stores it verbatim (no synthesized/duplicate
/// `_gar_id`), so the stored value byte-matches the UNWIND graph. Node groups
/// are passed in `LABELS` order, so the encoder's cumulative base offset for
/// label `li` equals `offsets[li]` — i.e. a node's global id == its `_gar_id`.
/// * *Edges* — the bulk encoder translates per-endpoint-label **0-based**
/// `_gar_id`s to global node ids by adding the group base. popviz's UNWIND edge
/// batches instead carry global `_gar_id`s (it MATCHes on the property), so for
/// the bulk lane we regenerate the edges with **zero offsets**
/// (`edge_batches(persons, 0, entities, 0)`): `src = j % persons`,
/// `dst = j` are exactly the 0-based positions the encoder expects, and
/// `base + position` reproduces the same global endpoints the UNWIND graph
/// MATCHed. Net: byte-identical nodes AND edges.
async fn load_falkor_bulk(
loader: &mut FalkorDbLoader,
state: &Arc<Mutex<PopvizState>>,
totals: &[i64],
offsets: &[i64],
t0: &Instant,
wake: &(impl Fn() + Send),
) -> Result<(), String> {
// sim: generate every label's vertex batches in parallel — fanning ROW-CHUNKS
// (not whole labels) across the gatling pool (LAW: gatling, never rayon), so
// the dominant label (60M-row TaxRecord) splits into many balanced units
// instead of stalling one core. Results are index-ordered, so `vbatches[li]`
// stays aligned to `LABELS` and carries the global `_gar_id` column the encoder
// reuses verbatim — byte-identical to the serial batches.
let vbatches = gen_all_vertex_batches(totals, offsets);
// Edge batches for the bulk lane: ZERO offsets → per-endpoint-label 0-based
// `_gar_id`s (see the fn doc). One `Person -> <label>` rel per non-Person label.
// `ebatches[li - 1]` = label `li` (index-ordered regroup over labels 1..).
let zero_offsets = vec![0i64; LABELS.len()];
let ebatches = gen_all_edge_batches(totals, 0, &zero_offsets);
let node_groups: Vec<BulkNodeGroup> = (0..LABELS.len())
.filter(|&li| !vbatches[li].is_empty())
.map(|li| BulkNodeGroup {
label: LABELS[li].0,
batches: &vbatches[li],
})
.collect();
let edge_groups: Vec<BulkEdgeGroup> = (1..LABELS.len())
.filter(|&li| !ebatches[li - 1].is_empty())
.map(|li| BulkEdgeGroup {
src_label: LABELS[0].0,
edge_label: PERSON_REL[li],
dst_label: LABELS[li].0,
batches: &ebatches[li - 1],
})
.collect();
// One ordered, appendable GRAPH.BULK stream: all node blobs, then all edge
// blobs. The server assigns global ids in blob order (== `LABELS` order ==
// popviz's offset order), so the produced graph matches the UNWIND path.
//
// The stream reports its CUMULATIVE (nodes, relations) after each blob lands, so
// the observable state advances *during* the load. It used to settle every label
// to 100% only after the whole call returned — which is why a UI bar driven off
// `fraction()` had exactly two reachable values, 0 and 1. Blobs are 1:1 with the
// groups and ordered by `LABELS`, so distributing the cumulative count over the
// label totals in order is exact, not an estimate.
let (nodes, edges) = loader
.bulk_insert_streamed_with_progress(&node_groups, &edge_groups, |n, r| {
{
let mut s = state.lock().unwrap();
let label_totals: Vec<usize> = s.labels.iter().map(|l| l.total).collect();
for (l, done) in s
.labels
.iter_mut()
.zip(distribute_progress(&label_totals, n))
{
l.loaded = done;
}
let rel_totals: Vec<usize> = s.rels.iter().map(|x| x.total).collect();
for (x, done) in s.rels.iter_mut().zip(distribute_progress(&rel_totals, r)) {
x.loaded = done;
}
s.elapsed_s = t0.elapsed().as_secs_f64();
}
wake();
})
.await
.map_err(|e| format!("bulk: {e}"))?;
// Settle the vertex counters from the REAL server-reported node count — NOT
// from the label totals. Force-setting `l.loaded = l.total` (a constant) and
// dropping `(nodes, edges)` outside debug builds painted every lane 100%/Done
// even when the stream no-op'd or the server silently rejected a blob (0 or
// partial rows landed). Distribute the ACTUAL loaded count over the totals,
// exactly like the per-callback honest settle the UNWIND / Iceberg lanes do.
{
let mut s = state.lock().unwrap();
let label_totals: Vec<usize> = s.labels.iter().map(|l| l.total).collect();
for (l, done) in s
.labels
.iter_mut()
.zip(distribute_progress(&label_totals, nodes))
{
l.loaded = done;
}
s.elapsed_s = t0.elapsed().as_secs_f64();
}
wake();
// Build the `_gar_id` indexes for parity with the UNWIND end-state and
// downstream query perf (the bulk edge resolution itself needs no index —
// endpoints are addressed directly, not MATCHed).
set_phase(state, Phase::Indexes);
wake();
for (name, _) in LABELS.iter() {
loader
.create_index(name, "_gar_id")
.await
.map_err(|e| format!("index {name}: {e}"))?;
}
// Settle the edge progress from the REAL server-reported relationship count
// (edges landed in the bulk stream above) — same honesty as the vertices.
set_phase(state, Phase::Edges);
{
let mut s = state.lock().unwrap();
let rel_totals: Vec<usize> = s.rels.iter().map(|x| x.total).collect();
for (x, done) in s
.rels
.iter_mut()
.zip(distribute_progress(&rel_totals, edges))
{
x.loaded = done;
}
s.elapsed_s = t0.elapsed().as_secs_f64();
}
wake();
// HONEST TERMINAL GATE. Only fall through to `Ok(())` — which lets `run_load`
// flip [`Phase::Done`] and emit a GREEN `functional_status("population_load",
// true)` — when FalkorDB actually reported EVERY expected node and edge landed.
// A no-op / partial / silently-rejected bulk stream (nodes or edges short of
// expected) is a FAILED load, NOT Done: return `Err` so `run_load` lands in
// [`Phase::Failed`] and emits `functional_status(ok=false)`. This is the real
// measurement the release build used to `let _`-discard.
bulk_load_verdict(nodes, edges, totals)?;
Ok(())
}
/// Honest terminal verdict for a streamed `GRAPH.BULK` load: the load may only be
/// reported GREEN/[`Phase::Done`] when the server-reported cumulative counts match
/// the expected population. `nodes`/`edges` are the REAL `(nodes, relations)` the
/// bulk stream returned; `expected_nodes == sum(totals)`,
/// `expected_edges == sum(totals[1..])` (one `Person -> <label>` rel per non-Person
/// label). A no-op (0/0) or partial stream is a genuine FAILURE — never silent Done.
fn bulk_load_verdict(nodes: usize, edges: usize, totals: &[i64]) -> Result<(), String> {
let expected_nodes = totals.iter().sum::<i64>() as usize;
let expected_edges = totals[1..].iter().sum::<i64>() as usize;
if nodes != expected_nodes || edges != expected_edges {
return Err(format!(
"bulk load shortfall: FalkorDB reported {nodes}/{expected_nodes} nodes, \
{edges}/{expected_edges} relationships landed — graph did not fully populate"
));
}
Ok(())
}
/// Insert ONE bounded window of a single label's vertex batches over the
/// multiplexed connection (up to `concurrency` UNWIND-CREATE queries in flight),
/// advancing the observable state per completed query. Splitting a label's batches
/// across several such windowed calls is graph-invariant — vertices are `CREATE`d
/// (order-independent) and each carries its global `_gar_id` — so the produced
/// graph is byte-identical to one big call, but peak memory stays a window.
async fn insert_vertex_window(
loader: &FalkorDbLoader,
li: usize,
batches: &[RecordBatch],
concurrency: usize,
state: &Arc<Mutex<PopvizState>>,
t0: &Instant,
wake: &(impl Fn() + Send),
) -> Result<(), String> {
loader
.insert_vertices_multi_par(&[LABELS[li].0], batches, concurrency, |n| {
advance(state, li, n, t0);
wake();
})
.await
.map_err(|e| format!("{}: {e}", LABELS[li].0))?;
Ok(())
}
/// Edge sibling of [`insert_vertex_window`] — insert ONE bounded window of the
/// `Person -[REL]-> LABELS[li]` relationship's batches (matched on `_gar_id`).
/// Order-independent (`MATCH` endpoints then `CREATE` a fresh rel), so windowing a
/// relationship across calls is byte-identical to one call.
async fn insert_edge_window(
loader: &FalkorDbLoader,
li: usize,
batches: &[RecordBatch],
concurrency: usize,
state: &Arc<Mutex<PopvizState>>,
t0: &Instant,
wake: &(impl Fn() + Send),
) -> Result<(), String> {
loader
.insert_edges_keyed_par(
LABELS[0].0,
PERSON_REL[li],
LABELS[li].0,
&["_gar_id"],
&["_gar_id"],
batches,
concurrency,
|n| {
advance_edges(state, li, n, t0);
wake();
},
)
.await
.map_err(|e| format!("edges {}: {e}", PERSON_REL[li]))?;
Ok(())
}
/// Parallel-UNWIND incremental lane — load every label's vertices into FalkorDB
/// (`UNWIND … CREATE`), then index `_gar_id`, then the edges. Multicore on BOTH
/// axes (the fix for the ~1-core ingest stall): the synthetic Arrow batches are
/// generated in PARALLEL across labels on the gatling pool (`sim`), and each label's
/// batches are inserted CONCURRENTLY over the multiplexed connection via
/// `insert_*_par` (`sink`, up to `available_parallelism()` queries in flight).
/// Invariants preserved exactly: vertices (a barrier) before edges; index
/// creation before edges. The produced graph is identical to the serial path —
/// `_gar_id` is carried in the batch column (global, offset-based) and
/// `CREATE`/`MATCH` are order-independent, so concurrency changes only dispatch
/// order, never content — and byte-identical to the bulk lane (see the parity test).
/// (Generation fans out on the **gatling** worker pool — LAW: gatling, never rayon.)
async fn load_falkor_unwind(
loader: &mut FalkorDbLoader,
state: &Arc<Mutex<PopvizState>>,
totals: &[i64],
offsets: &[i64],
t0: &Instant,
wake: &(impl Fn() + Send),
) -> Result<(), String> {
// Up to this many UNWIND-CREATE queries in flight per label (one clone of the
// multiplexed conn each). ~11–12 on Loki — where a serial link sat at ~1.
let concurrency = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
// Drive the whole load off the canonical phase order (vertices → indexes →
// edges). Indexing BEFORE the edge phase is the key throughput fix: each edge
// MATCHes its endpoints by `_gar_id`, so without the index every MATCH is an
// O(V) label scan — a quadratic SERVER-side stall the client-side pipelining
// can't reach. See [`falkor_load_phase_order`] for the full reasoning; it is
// the single source of truth for the ordering (a unit test guards it).
for phase in falkor_load_phase_order() {
match phase {
// sim + sink: STREAM every label's vertex batches through a bounded
// [`BatchStream`] (gatling row-chunk fan-out — the biggest single-core
// chunk — but only one window held at a time), inserting each label's
// batches concurrently in windows as they arrive instead of collecting
// the whole population first (the OOM fix: peak memory is a window). A
// label's batches arrive contiguously (plan order), so we accumulate up
// to a window and flush on the label boundary or when the window fills;
// per-label await keeps the vertices-before-{index,edges} barrier.
Phase::Vertices => {
let schemas: Vec<Arc<Schema>> =
LABELS.iter().map(|(l, _)| vertex_schema(l)).collect();
let vplan = vertex_plan(totals);
let vgen = |li: usize, start: i64, end: i64| {
vertex_batch_range(LABELS[li].0, &schemas[li], start, end, offsets[li])
};
let win = stream_window();
let mut cur: Option<usize> = None;
let mut pending: Vec<RecordBatch> = Vec::new();
for (li, b) in BatchStream::new(&vplan, vgen, win) {
if let Some(d) = cur {
if d != li || pending.len() >= win {
insert_vertex_window(loader, d, &pending, concurrency, state, t0, wake)
.await?;
pending.clear();
}
}
cur = Some(li);
pending.push(b);
}
if let Some(d) = cur {
if !pending.is_empty() {
insert_vertex_window(loader, d, &pending, concurrency, state, t0, wake)
.await?;
}
}
}
// Build the `_gar_id` index on every label BEFORE any edge MATCHes on
// it — turns the edge phase's endpoint lookups from O(V) label scans
// into indexed O(1) lookups. Vertices are already loaded (prior phase).
Phase::Indexes => {
set_phase(state, Phase::Indexes);
wake();
for (name, _) in LABELS.iter() {
loader
.create_index(name, "_gar_id")
.await
.map_err(|e| format!("index {name}: {e}"))?;
}
}
// Edges: one `Person -[REL]-> <label>` relationship per non-Person
// label, matched on `_gar_id` (now indexed). STREAM the edge plan the
// same bounded way — accumulate a relationship's batches (contiguous in
// plan order) up to a window and flush on the rel boundary / full
// window, so peak memory stays a window, not every edge batch at once.
Phase::Edges => {
set_phase(state, Phase::Edges);
wake();
let eplan = edge_plan(totals);
let person_count = totals[0];
let person_offset = offsets[0];
let egen = |li: usize, start: i64, end: i64| {
edge_batch_range(person_count, person_offset, offsets[li], start, end)
};
let win = stream_window();
let mut cur: Option<usize> = None;
let mut pending: Vec<RecordBatch> = Vec::new();
for (li, b) in BatchStream::new(&eplan, egen, win) {
if let Some(d) = cur {
if d != li || pending.len() >= win {
insert_edge_window(loader, d, &pending, concurrency, state, t0, wake)
.await?;
pending.clear();
}
}
cur = Some(li);
pending.push(b);
}
if let Some(d) = cur {
if !pending.is_empty() {
insert_edge_window(loader, d, &pending, concurrency, state, t0, wake)
.await?;
}
}
}
// The load order only ever yields the three data phases above.
Phase::Idle | Phase::Done | Phase::Failed => {}
}
}
Ok(())
}
/// Load every label's vertices into Iceberg tables — one table per label, then a
/// `rel_<REL>` per relationship — in `lake`: a local skade warehouse or an Iceberg REST
/// catalog ([`lake::Lake`]). One append commit per batch; zero-copy (knut and skade share
/// arrow 59, so the batch hands straight to iceberg + skade).
#[cfg(feature = "iceberg")]
async fn load_iceberg(
lake: &lake::Lake,
drop: bool,
state: &Arc<Mutex<PopvizState>>,
totals: &[i64],
offsets: &[i64],
t0: &Instant,
wake: &(impl Fn() + Send),
) -> Result<(), String> {
if let (true, lake::Lake::Dir(dir)) = (drop, lake) {
if dir.exists() {
// Same rule as the Falkor path: `--drop` wipes the whole warehouse, so
// never do it silently - announce it before removing the directory.
eprintln!("knut-popviz: --drop removing existing warehouse {dir:?} before load");
let _ = std::fs::remove_dir_all(dir);
}
}
let open = lake.open().await?;
if drop {
// A catalog lake: the tables this load writes and nothing else, announced the same way.
let ours: Vec<String> = LABELS.iter().map(|(l, _)| l.to_string()).chain(PERSON_REL.iter().skip(1).map(|r| format!("rel_{r}"))).collect();
let n = open.drop_tables(&ours).await?;
if n > 0 {
eprintln!("knut-popviz: --drop dropped {n} tables of {} before load", lake.describe());
}
}
// STREAM every label's vertex batches through a bounded [`BatchStream`] (LAW
// #0: gatling, never rayon) — fanning ROW-CHUNKS across cores (the CPU-bound
// Arrow gen was the ~1-core serial stall) but holding only ONE window of
// batches at a time, appending + dropping each before the next window is
// generated (the fix for the OOM: peak memory is a window, not the whole
// graph). The Iceberg append itself stays ORDERED per table (each label is one
// table, plan order groups a label's batches contiguously — a single writer).
let schemas: Vec<Arc<Schema>> = LABELS.iter().map(|(l, _)| vertex_schema(l)).collect();
let vplan = vertex_plan(totals);
let vgen = |li: usize, start: i64, end: i64| {
vertex_batch_range(LABELS[li].0, &schemas[li], start, end, offsets[li])
};
for (li, batch) in BatchStream::new(&vplan, vgen, stream_window()) {
let n = open.append(LABELS[li].0, with_round(batch, 0)).await?;
advance(state, li, n, t0);
wake();
}
// Edges: one Iceberg table per `Person -> <label>` relationship (named after
// the rel), carrying the `src`/`dst` `_gar_id` endpoint columns. Streamed the
// same bounded way — plan order groups a rel's batches contiguously, so each
// `(li, batch)` appends to `rel_<REL>` in order (byte-identical to buffered).
set_phase(state, Phase::Edges);
wake();
let eplan = edge_plan(totals);
let person_count = totals[0];
let person_offset = offsets[0];
let egen = |li: usize, start: i64, end: i64| {
edge_batch_range(person_count, person_offset, offsets[li], start, end)
};
for (li, batch) in BatchStream::new(&eplan, egen, stream_window()) {
let table = format!("rel_{}", PERSON_REL[li]);
let n = open.append(&table, with_round(batch, 0)).await.map_err(|e| format!("edges {e}"))?;
advance_edges(state, li, n, t0);
wake();
}
Ok(())
}
/// `batch` with a trailing `round` Int64 column, every row `round`. The Iceberg target
/// carries it on every vertex and `rel_` table so [`permutate`] can stamp what it changed
/// and a viewer can light up one round's changes. The FalkorDB target does not carry it.
#[cfg(feature = "iceberg")]
pub(crate) fn with_round(batch: RecordBatch, round: i64) -> RecordBatch {
let mut fields: Vec<Field> = batch.schema().fields().iter().map(|f| f.as_ref().clone()).collect();
fields.push(Field::new("round", DataType::Int64, false));
let mut cols = batch.columns().to_vec();
cols.push(Arc::new(Int64Array::from(vec![round; batch.num_rows()])));
RecordBatch::try_new(Arc::new(Schema::new(fields)), cols).expect("round column")
}
#[cfg(not(feature = "iceberg"))]
async fn load_iceberg(
_lake: &lake::Lake,
_drop: bool,
_state: &Arc<Mutex<PopvizState>>,
_totals: &[i64],
_offsets: &[i64],
_t0: &Instant,
_wake: &(impl Fn() + Send),
) -> Result<(), String> {
Err("Iceberg target needs a build with `--features iceberg`".into())
}
fn advance(state: &Arc<Mutex<PopvizState>>, li: usize, n: usize, t0: &Instant) {
let mut s = state.lock().unwrap();
s.labels[li].loaded += n;
s.elapsed_s = t0.elapsed().as_secs_f64();
}
/// Record `n` loaded edges for the `Person -> LABELS[li]` relationship. `rels`
/// is indexed by destination label (`li - 1`, since `Person`/li 0 has no rel).
fn advance_edges(state: &Arc<Mutex<PopvizState>>, li: usize, n: usize, t0: &Instant) {
let mut s = state.lock().unwrap();
s.rels[li - 1].loaded += n;
s.elapsed_s = t0.elapsed().as_secs_f64();
}
fn set_phase(state: &Arc<Mutex<PopvizState>>, phase: Phase) {
state.lock().unwrap().phase = phase;
}
fn fail(state: &Arc<Mutex<PopvizState>>, msg: String) {
let mut s = state.lock().unwrap();
s.phase = Phase::Failed;
s.error = Some(msg);
}
#[cfg(test)]
mod tests {
use super::*;
/// **RED-when-broken: the streamed bulk lane reports PARTIAL progress.** The
/// `GRAPH.BULK` fast lane (auto-selected for every fresh/empty graph — i.e. every
/// zone generate) used to settle every label to 100% only *after* the whole
/// population had landed, so a UI bar driven off [`PopvizState::fraction`] had
/// exactly two reachable values: 0 and 1. Feeding the loader's cumulative
/// per-blob counter through [`distribute_progress`] must produce a fraction that
/// is STRICTLY between 0 and 1 at the intermediate points. If the per-blob
/// reporting were removed (back to a single settle), every observed fraction
/// would be 0 or 1 and this is red.
#[test]
fn streamed_bulk_progress_is_partial_not_just_0_and_1() {
let mut st = PopvizState::new(1e-6);
let totals: Vec<usize> = st.labels.iter().map(|l| l.total).collect();
let grand: usize = totals.iter().sum();
assert!(
grand > 2,
"the demo population has enough nodes to show progress"
);
// Walk the cumulative counter the way the streamed lane reports it (one
// callback per node blob == per label group, in LABELS order).
let mut fractions = Vec::new();
let mut cum = 0usize;
for &t in &totals {
cum += t;
for (l, done) in st.labels.iter_mut().zip(distribute_progress(&totals, cum)) {
l.loaded = done;
}
fractions.push(st.fraction());
}
// The end state is exactly 100% …
assert_eq!(
*fractions.last().unwrap(),
1.0,
"the last blob completes the load"
);
// … and at least one report before it was a real, partial value.
let partial: Vec<f32> = fractions
.iter()
.copied()
.filter(|f| *f > 0.0 && *f < 1.0)
.collect();
assert!(
!partial.is_empty(),
"the streamed bulk lane must report progress BETWEEN 0 and 1, got {fractions:?}",
);
// Progress never goes backwards.
for w in fractions.windows(2) {
assert!(w[1] >= w[0], "bulk progress is monotonic: {fractions:?}");
}
functional_status(
"knut-popviz/bulk_progress",
"streamed_bulk_reports_partial_progress",
!partial.is_empty(),
"GRAPH.BULK fast lane advances PopvizState per blob (0<f<1), not one settle to 100%",
);
}
/// **RED-when-broken: the bulk lane must NOT report Done/green over a no-op or
/// partial load.** This is the CRITICAL false-green the fix closes. The streamed
/// `GRAPH.BULK` lane used to force every counter to its total and `let _`-discard
/// the server-reported `(nodes, edges)` in release, so a bulk stream that landed
/// 0 rows (server silently rejected every blob) still flipped [`Phase::Done`] +
/// `functional_status("population_load", true)`. The terminal verdict must instead
/// be driven by the REAL landed counts: 0/0 (or any shortfall) is `Err` → the
/// caller lands in [`Phase::Failed`], and only the fully-landed population is `Ok`.
#[test]
fn bulk_load_noop_is_failed_not_done() {
let st = PopvizState::new(1e-6);
let totals: Vec<i64> = st.labels.iter().map(|l| l.total as i64).collect();
let expected_nodes = totals.iter().sum::<i64>() as usize;
let expected_edges = totals[1..].iter().sum::<i64>() as usize;
assert!(
expected_nodes > 0 && expected_edges > 0,
"demo population is non-empty"
);
// RED axis: a no-op bulk stream (0 nodes, 0 edges landed) must be a FAILURE.
// Before the fix this path force-set every counter to total and reported Done.
assert!(
bulk_load_verdict(0, 0, &totals).is_err(),
"a bulk load that landed 0/0 rows must NOT be reported Done/green",
);
// A partial load (nodes short, or edges short) is also a failure, not Done.
assert!(
bulk_load_verdict(expected_nodes - 1, expected_edges, &totals).is_err(),
"a node shortfall must fail",
);
assert!(
bulk_load_verdict(expected_nodes, expected_edges - 1, &totals).is_err(),
"an edge shortfall must fail",
);
// GREEN axis: the full, real population landing is the ONLY Ok/Done state.
assert!(
bulk_load_verdict(expected_nodes, expected_edges, &totals).is_ok(),
"the fully-landed population is the only green terminal state",
);
functional_status(
"knut-popviz/bulk_load",
"bulk_done_gated_on_real_landed_counts",
bulk_load_verdict(0, 0, &totals).is_err()
&& bulk_load_verdict(expected_nodes, expected_edges, &totals).is_ok(),
"GRAPH.BULK Done/green requires server-reported nodes+edges == expected (no-op reds)",
);
}
/// [`distribute_progress`] is an EXACT prefix fill: full groups first, at most one
/// partial group at the frontier, and it never exceeds a group's total.
#[test]
fn distribute_progress_fills_groups_in_order() {
let totals = [3usize, 2, 4];
assert_eq!(distribute_progress(&totals, 0), vec![0, 0, 0]);
assert_eq!(
distribute_progress(&totals, 2),
vec![2, 0, 0],
"partial first group"
);
assert_eq!(
distribute_progress(&totals, 3),
vec![3, 0, 0],
"first group complete"
);
assert_eq!(distribute_progress(&totals, 5), vec![3, 2, 0]);
assert_eq!(
distribute_progress(&totals, 7),
vec![3, 2, 2],
"partial last group"
);
assert_eq!(
distribute_progress(&totals, 9),
vec![3, 2, 4],
"all complete"
);
// Over-reporting clamps at the totals (never above 100%).
assert_eq!(distribute_progress(&totals, 99), vec![3, 2, 4]);
// The sum always equals min(done, grand total) — nothing is lost or invented.
for done in 0..12 {
let got: usize = distribute_progress(&totals, done).iter().sum();
assert_eq!(
got,
done.min(9),
"distribution conserves the count at done={done}"
);
}
}
/// **Byte-identity proof for the row-chunk fan-out.** The parallel
/// [`gen_all_vertex_batches`] / [`gen_all_edge_batches`] (which split each
/// label's rows into balanced ≤`CHUNK` units across cores) must produce the
/// EXACT same `Vec<Vec<RecordBatch>>` — same batch boundaries, same order, same
/// cell values — as the old whole-label serial `vertex_batches`/`edge_batches`
/// generation. RED-when-broken: a wrong per-chunk offset (`_gar_id`), a dropped
/// or reordered chunk, or a changed `CHUNK` boundary diverges a batch here.
#[test]
fn gen_all_matches_serial_byte_identical() {
let s = PopvizState::new(0.001);
let totals: Vec<i64> = s.labels.iter().map(|l| l.total as i64).collect();
let mut offsets = vec![0i64; totals.len()];
let mut acc = 0i64;
for (i, c) in totals.iter().enumerate() {
offsets[i] = acc;
acc += c;
}
// Vertices: parallel row-chunk fan-out vs serial per-label generation.
let par_v = gen_all_vertex_batches(&totals, &offsets);
let ser_v: Vec<Vec<RecordBatch>> = (0..LABELS.len())
.map(|li| vertex_batches(LABELS[li].0, totals[li], offsets[li]))
.collect();
assert_eq!(
par_v, ser_v,
"parallel vertex batches must be byte-identical to serial"
);
// Edges — UNWIND/Iceberg offsets (real cumulative).
let par_e = gen_all_edge_batches(&totals, offsets[0], &offsets);
let ser_e: Vec<Vec<RecordBatch>> = (1..LABELS.len())
.map(|li| edge_batches(totals[0], offsets[0], totals[li], offsets[li]))
.collect();
assert_eq!(
par_e, ser_e,
"parallel edge batches must be byte-identical to serial (real offsets)"
);
// Edges — bulk lane (zero offsets).
let zeros = vec![0i64; LABELS.len()];
let par_e0 = gen_all_edge_batches(&totals, 0, &zeros);
let ser_e0: Vec<Vec<RecordBatch>> = (1..LABELS.len())
.map(|li| edge_batches(totals[0], 0, totals[li], 0))
.collect();
assert_eq!(
par_e0, ser_e0,
"parallel edge batches must be byte-identical to serial (zero offsets)"
);
}
/// **Bounded-stream byte-identity proof.** The whole point of [`BatchStream`]
/// is to hand batches to the loader incrementally (peak memory = one window)
/// instead of the buffered [`gen_all_vertex_batches`]/[`gen_all_edge_batches`]
/// collecting the entire population first. This asserts that draining a
/// **small-window** stream (window = 2/3 — deliberately far below the whole
/// plan, so it regenerates many windows) reproduces the buffered whole-graph
/// output EXACTLY: same batch boundaries, same order, same cell values. That is
/// the streamed-output-equals-collected-reference contract — a bulk-loaded /
/// UNWIND / Iceberg graph is byte-identical whether generated streaming or
/// buffered. RED-when-broken: a wrong window cursor, a dropped/reordered/
/// duplicated window batch, or a per-window offset error diverges a batch here.
#[test]
fn stream_matches_buffered_byte_identical() {
let s = PopvizState::new(0.001);
let totals: Vec<i64> = s.labels.iter().map(|l| l.total as i64).collect();
let mut offsets = vec![0i64; totals.len()];
let mut acc = 0i64;
for (i, c) in totals.iter().enumerate() {
offsets[i] = acc;
acc += c;
}
// Vertices: buffered whole-graph collect vs a tiny-window incremental drain.
let buffered_v = gen_all_vertex_batches(&totals, &offsets);
let schemas: Vec<Arc<Schema>> = LABELS.iter().map(|(l, _)| vertex_schema(l)).collect();
let vplan = vertex_plan(&totals);
let vgen = |li: usize, start: i64, end: i64| {
vertex_batch_range(LABELS[li].0, &schemas[li], start, end, offsets[li])
};
let mut streamed_v: Vec<Vec<RecordBatch>> = (0..LABELS.len()).map(|_| Vec::new()).collect();
for (li, b) in BatchStream::new(&vplan, vgen, 2) {
streamed_v[li].push(b);
}
assert_eq!(
streamed_v, buffered_v,
"streamed vertex batches (window=2) must be byte-identical to the buffered whole-graph generation"
);
// Edges: same, with the real cumulative offsets (UNWIND/Iceberg lanes).
let buffered_e = gen_all_edge_batches(&totals, offsets[0], &offsets);
let eplan = edge_plan(&totals);
let person_count = totals[0];
let person_offset = offsets[0];
let egen = |li: usize, start: i64, end: i64| {
edge_batch_range(person_count, person_offset, offsets[li], start, end)
};
let mut streamed_e: Vec<Vec<RecordBatch>> =
(0..LABELS.len() - 1).map(|_| Vec::new()).collect();
for (li, b) in BatchStream::new(&eplan, egen, 3) {
streamed_e[li - 1].push(b);
}
assert_eq!(
streamed_e, buffered_e,
"streamed edge batches (window=3) must be byte-identical to the buffered generation"
);
}
/// **Core-saturation bench** (ignored by default — run with
/// `cargo test -p knut-popviz --features iceberg -- --ignored --nocapture
/// bench_core_saturation`). Generates a ~1.2M-node population (scale 0.01) two
/// ways and reports `cores_busy = cpu_time / wall_time` (CPU time = this
/// process's `utime+stime` across all threads, from `/proc/self/stat`):
/// * OLD: one gatling unit per whole label (12 uneven units — the dominant
/// 60M-row `TaxRecord` stalls one core, ~35% box utilization);
/// * NEW: [`gen_all_vertex_batches`] — one balanced ≤`CHUNK` row-range per
/// unit (~49 units at this scale), saturating all cores.
#[test]
#[ignore]
fn bench_core_saturation() {
use std::time::Instant;
use znippy_zoomies::gatling_forkjoin::{default_workers, gatling_for_each};
let scale: f64 = std::env::var("KNUT_BENCH_SCALE")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(0.01);
let s = PopvizState::new(scale);
let totals: Vec<i64> = s.labels.iter().map(|l| l.total as i64).collect();
let mut offsets = vec![0i64; totals.len()];
let mut acc = 0i64;
for (i, c) in totals.iter().enumerate() {
offsets[i] = acc;
acc += c;
}
let nodes: i64 = totals.iter().sum();
// Read this process's cumulative CPU jiffies (utime+stime, all threads).
fn cpu_jiffies() -> u64 {
let stat = std::fs::read_to_string("/proc/self/stat").unwrap();
// Field layout after the (comm) paren: fields 14 (utime) & 15 (stime),
// 1-based per proc(5). Split on the last ')' to skip a spaced comm.
let after = &stat[stat.rfind(')').unwrap() + 2..];
let f: Vec<&str> = after.split_whitespace().collect();
// after[0] = state (field 3); utime = field 14 => index 11, stime => 12.
f[11].parse::<u64>().unwrap() + f[12].parse::<u64>().unwrap()
}
let clk_tck = 100u64; // CONFIG_HZ=100 on this box; jiffies→s divisor.
let cores = default_workers();
// Repeat each variant so the measurement window is ~0.5s+ (the 100Hz CPU
// clock is too coarse for a single ~25ms generation) — peak memory is one
// population's transient Arrow, unchanged by the repeat.
const ITERS: usize = 30;
let run = |label: &str, f: &dyn Fn()| {
let c0 = cpu_jiffies();
let w0 = Instant::now();
for _ in 0..ITERS {
f();
}
let wall = w0.elapsed().as_secs_f64();
let cpu = (cpu_jiffies() - c0) as f64 / clk_tck as f64;
let busy = cpu / wall;
eprintln!(
"[{label}] {ITERS}x wall={wall:.3}s cpu={cpu:.3}s cores_busy={busy:.1}/{cores} ({:.0}% of box)",
100.0 * busy / cores as f64
);
};
eprintln!(
"scale={scale} nodes={nodes} labels={} cores={cores}",
LABELS.len()
);
// OLD: one gatling work-unit per whole label (the ~35% stall).
run("OLD whole-label", &|| {
let v: Vec<Vec<RecordBatch>> = gatling_for_each(LABELS.len(), 0, |li| {
vertex_batches(LABELS[li].0, totals[li], offsets[li])
});
std::hint::black_box(v);
});
// NEW: balanced row-chunk fan-out.
run("NEW row-chunk", &|| {
let v = gen_all_vertex_batches(&totals, &offsets);
std::hint::black_box(v);
});
}
/// The FalkorDB load must build the `_gar_id` indexes AFTER the vertices but
/// BEFORE the edge phase — otherwise every edge `MATCH (a {_gar_id: …})` is an
/// unindexed O(V) label scan and the edge phase becomes O(edges × vertices)
/// server-side work (the ingest stall). `load_falkor` drives its phase
/// sequence off `falkor_load_phase_order`, so asserting the order here — with
/// no live DB — proves the load actually indexes before it matches endpoints.
/// RED-when-broken: revert to indexes-after-edges (`ii > ei`) and this fails.
#[test]
fn falkor_indexes_are_built_before_edges() {
let order = falkor_load_phase_order();
let vi = order
.iter()
.position(|p| *p == Phase::Vertices)
.expect("vertices phase present");
let ii = order
.iter()
.position(|p| *p == Phase::Indexes)
.expect("indexes phase present");
let ei = order
.iter()
.position(|p| *p == Phase::Edges)
.expect("edges phase present");
assert!(
vi < ii,
"vertices must be inserted before the _gar_id index is built over them"
);
assert!(
ii < ei,
"the _gar_id index must be built BEFORE the edge phase so endpoint \
MATCHes are indexed O(1) lookups, not O(V) label scans (the stall)"
);
assert_eq!(
order.len(),
3,
"no data phase silently dropped from the load"
);
}
#[test]
fn state_json_reflects_progress() {
let mut s = PopvizState::new(0.0001); // tiny totals
let json = s.state_json();
assert_eq!(json["phase"], "idle");
assert_eq!(json["loaded"], 0);
assert_eq!(json["labels"].as_object().unwrap().len(), 12);
assert!(s.total() > 0, "scaled totals are non-zero");
assert!(
json["placeholders"]
.as_array()
.unwrap()
.iter()
.any(|p| p == "not started"),
"idle → 'not started' placeholder"
);
// simulate some progress
s.phase = Phase::Vertices;
s.labels[0].loaded = s.labels[0].total;
s.elapsed_s = 1.0;
let json = s.state_json();
assert_eq!(json["phase"], "loading vertices");
assert_eq!(json["loaded"], s.labels[0].total);
assert!(json["fraction"].as_f64().unwrap() > 0.0);
assert!(json["placeholders"].as_array().unwrap().is_empty());
}
#[test]
fn person_batches_carry_swedish_geo_and_ekonomiskt_bistand() {
use arrow_array::{Array, BooleanArray, Float64Array, StringArray};
let batches = vertex_batches("Person", 500, 0);
assert!(!batches.is_empty());
let b = &batches[0];
// geo + flag columns are present on Person.
for col in [
"_gar_id",
"idx",
"name",
"lon",
"lat",
"kommun",
"ekonomiskt_bistand",
] {
assert!(
b.column_by_name(col).is_some(),
"Person batch has `{col}` column"
);
}
let (mnx, mny, mxx, mxy) = geo::SWEDEN_BBOX;
let lon = b
.column_by_name("lon")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let lat = b
.column_by_name("lat")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let kommun = b
.column_by_name("kommun")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let flag = b
.column_by_name("ekonomiskt_bistand")
.unwrap()
.as_any()
.downcast_ref::<BooleanArray>()
.unwrap();
let mut recipients = 0;
for i in 0..b.num_rows() {
assert!(
lon.value(i) >= mnx && lon.value(i) <= mxx,
"row {i} lon in Sweden"
);
assert!(
lat.value(i) >= mny && lat.value(i) <= mxy,
"row {i} lat in Sweden"
);
assert!(!kommun.value(i).is_empty(), "row {i} has a kommun");
if flag.value(i) {
recipients += 1;
}
}
assert!(recipients > 0, "some Persons are ekonomiskt_bistand");
}
#[test]
fn non_geo_labels_stay_lean() {
// A non-geo label (e.g. Loan) keeps the original 3-column schema.
let batches = vertex_batches("Loan", 100, 0);
let b = &batches[0];
assert!(b.column_by_name("lon").is_none(), "Loan carries no geo");
assert_eq!(b.num_columns(), 3, "_gar_id/idx/name only");
}
#[test]
fn label_totals_scale() {
let s = PopvizState::new(0.001);
// Person is the biggest label at 11M → 11k at scale 0.001.
assert_eq!(s.labels[0].name, "Person");
assert_eq!(s.labels[0].total, 11_000);
}
#[test]
fn edge_batches_key_src_dst_on_gar_id() {
use arrow_array::{Array, Int64Array};
// Person ids live in [1000, 1004); entity ids in [5000, 5010).
let person_count = 4i64;
let person_offset = 1000i64;
let entity_count = 10i64;
let entity_offset = 5000i64;
let batches = edge_batches(person_count, person_offset, entity_count, entity_offset);
assert!(!batches.is_empty(), "some edge batches produced");
// Exactly the `src`/`dst` endpoint key columns, both Int64 `_gar_id`s.
let b = &batches[0];
let schema = b.schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(names, vec!["src", "dst"], "keyed on _gar_id via src/dst");
// One edge per entity, deterministic: dst = entity_offset + j, and
// src = person_offset + (j % person_count) — every dst is a real entity
// id and every src is a real person id.
let total: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(total, entity_count as usize, "one edge per entity");
let src = b.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
let dst = b.column(1).as_any().downcast_ref::<Int64Array>().unwrap();
for row in 0..b.num_rows() {
let j = row as i64;
assert_eq!(dst.value(row), entity_offset + j);
assert_eq!(src.value(row), person_offset + (j % person_count));
assert!((person_offset..person_offset + person_count).contains(&src.value(row)));
}
}
#[test]
fn state_json_reports_edge_progress() {
let mut s = PopvizState::new(0.0001);
// One rel per non-Person label (11 of the 12 labels).
assert_eq!(s.rels.len(), LABELS.len() - 1);
assert_eq!(s.rels[0].name, PERSON_REL[1]); // Person -> Address = LIVES_AT
assert_eq!(s.rels[0].dst_label, LABELS[1].0);
let json = s.state_json();
assert_eq!(json["edges_loaded"], 0, "nothing loaded at idle");
assert!(
json["edges_total"].as_u64().unwrap() > 0,
"edges are planned"
);
let edges = json["edges"].as_object().unwrap();
assert_eq!(edges.len(), LABELS.len() - 1, "one entry per relationship");
assert_eq!(edges[PERSON_REL[1]]["src"], "Person");
assert_eq!(edges[PERSON_REL[1]]["dst"], LABELS[1].0);
// Simulate the Edges phase making progress on one relationship.
s.phase = Phase::Edges;
s.rels[0].loaded = s.rels[0].total;
let json = s.state_json();
assert_eq!(json["phase"], "loading edges");
assert!(
json["edges_loaded"].as_u64().unwrap() > 0,
"state_json reports non-zero edges_loaded once the edge phase runs"
);
assert_eq!(json["edges"][PERSON_REL[1]]["loaded"], s.rels[0].total);
}
/// **DB-free reconciliation proof for the bulk fast lane.** Builds the SAME
/// `GRAPH.BULK` node/edge groups `load_falkor_bulk` feeds the loader and
/// asserts, with no server, that popviz's global-offset `_gar_id` scheme is
/// reconciled with the loader-level bulk encoder:
/// 1. the node blob for a non-first label carries its **global** `_gar_id`
/// (offset-based) and carries `_gar_id` **exactly once** (the dedup —
/// no synthesized second copy); and
/// 2. an edge built with ZERO offsets resolves to the SAME global endpoints
/// the UNWIND path MATCHes (`base + position`).
///
/// RED-when-broken: if the encoder re-synthesized `_gar_id`, the header would
/// gain a duplicate and the stored id would flip to 0-based; if the bulk edge
/// batches kept popviz's real offsets, the dst would double-count the base.
#[test]
fn bulk_groups_reconcile_gar_id_and_edges() {
// Totals/offsets computed exactly as `run_load` does (cumulative in LABELS
// order), so a group's base offset == `offsets[li]`.
let s = PopvizState::new(0.0002);
let totals: Vec<i64> = s.labels.iter().map(|l| l.total as i64).collect();
let mut offsets = vec![0i64; totals.len()];
let mut acc = 0i64;
for (i, c) in totals.iter().enumerate() {
offsets[i] = acc;
acc += c;
}
assert!(
totals[1] > 0 && offsets[1] > 0,
"Address is non-empty and offset past Person"
);
// The bulk lane's batches: vertices carry the global `_gar_id` column;
// edges use ZERO offsets (per-endpoint-label 0-based).
let vbatches: Vec<Vec<RecordBatch>> = (0..LABELS.len())
.map(|li| vertex_batches(LABELS[li].0, totals[li], offsets[li]))
.collect();
let ebatches: Vec<Vec<RecordBatch>> = (1..LABELS.len())
.map(|li| edge_batches(totals[0], 0, totals[li], 0))
.collect();
let node_groups: Vec<BulkNodeGroup> = (0..LABELS.len())
.map(|li| BulkNodeGroup {
label: LABELS[li].0,
batches: &vbatches[li],
})
.collect();
let edge_groups: Vec<BulkEdgeGroup> = (1..LABELS.len())
.map(|li| BulkEdgeGroup {
src_label: LABELS[0].0,
edge_label: PERSON_REL[li],
dst_label: LABELS[li].0,
batches: &ebatches[li - 1],
})
.collect();
let payload = graphar_falkordb::build_bulk_payload(&node_groups, &edge_groups)
.expect("popviz bulk payload encodes");
assert_eq!(
payload.node_count,
totals.iter().sum::<i64>() as u64,
"every vertex"
);
assert_eq!(
payload.relation_count,
totals[1..].iter().sum::<i64>() as u64,
"one edge per non-Person entity"
);
assert_eq!(
payload.node_blob_len,
LABELS.len(),
"one node blob per label"
);
// (1) The Address (li 1) node blob: header lists `_gar_id` exactly once,
// and the first node's stored `_gar_id` is the GLOBAL value `offsets[1]`.
let (hdr_props, first_gar) = parse_node_blob_head(&payload.blobs[1]);
assert_eq!(
hdr_props.iter().filter(|p| *p == "_gar_id").count(),
1,
"exactly one `_gar_id` in the header (dedup): {hdr_props:?}"
);
assert_eq!(
first_gar, offsets[1],
"Address node 0 stores its GLOBAL _gar_id"
);
// (2) The Person->Address edge blob (first relation blob, right after the
// node blobs): its first endpoint pair resolves to global ids — dst is
// `offsets[1] + 0`, src is a real Person global id in `[0, totals[0])`.
let (src0, dst0) = parse_edge_blob_first_endpoints(&payload.blobs[payload.node_blob_len]);
assert_eq!(
dst0, offsets[1] as u64,
"edge dst resolved to Address global id"
);
assert!(src0 < totals[0] as u64, "edge src is a Person global id");
}
/// Parse a node blob's header — returns (property names, first node's first
/// value as i64, which for popviz batches is the `_gar_id` LONG). Mirrors the
/// wire in `graphar_falkordb::bulk`.
fn parse_node_blob_head(blob: &[u8]) -> (Vec<String>, i64) {
let mut p = 0usize;
let read_cstr = |b: &[u8], p: &mut usize| {
let start = *p;
while b[*p] != 0 {
*p += 1;
}
let s = String::from_utf8_lossy(&b[start..*p]).into_owned();
*p += 1; // NUL
s
};
let _label = read_cstr(blob, &mut p);
let count = u32::from_le_bytes(blob[p..p + 4].try_into().unwrap()) as usize;
p += 4;
let props: Vec<String> = (0..count).map(|_| read_cstr(blob, &mut p)).collect();
// First node, first value: type byte then i64-LE (LONG) for `_gar_id`.
assert_eq!(blob[p], 4, "first value is a LONG (_gar_id)");
p += 1;
let gar = i64::from_le_bytes(blob[p..p + 8].try_into().unwrap());
(props, gar)
}
/// Parse a (no-prop) edge blob's first `(u64 src, u64 dst)` endpoint pair.
fn parse_edge_blob_first_endpoints(blob: &[u8]) -> (u64, u64) {
let mut p = 0usize;
while blob[p] != 0 {
p += 1; // reltype cstr
}
p += 1; // NUL
let prop_count = u32::from_le_bytes(blob[p..p + 4].try_into().unwrap());
p += 4;
assert_eq!(prop_count, 0, "popviz edges carry no properties");
let src = u64::from_le_bytes(blob[p..p + 8].try_into().unwrap());
p += 8;
let dst = u64::from_le_bytes(blob[p..p + 8].try_into().unwrap());
(src, dst)
}
}