crates/graphar-falkordb/src/loader.rs
use std::fmt::Write as _;
use arrow_array::{Array, RecordBatch};
use crate::{
bulk::{BulkEdgeGroup, BulkNodeGroup, build_bulk_payload},
cypher::{
ColRenderer, EdgeRow, arrow_value_to_cypher, render_labels, unwind_edge_query_keyed_param,
write_row_map,
},
error::{FalkorError, Result},
};
/// Default rows per `GRAPH.QUERY` call.
pub const DEFAULT_BATCH_SIZE: usize = 512;
/// Which `GRAPH.QUERY` statistic counts as "the load happened" for a phase.
///
/// The vertex phase is `CREATE`-only so its rows and its nodes agree; the edge
/// phase is `MATCH … CREATE`, so a row whose endpoints do not match produces a
/// **successful query that creates nothing**. Counting rows sent therefore
/// reports a full load over an empty graph, which is the exact live regression
/// (`korp_demo`: 121 200 nodes, 0 edges, GREEN status row).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StatKey {
NodesCreated,
RelationshipsCreated,
}
impl StatKey {
/// The statistics line prefix FalkorDB/RedisGraph emits for this entity.
fn needle(self) -> &'static str {
match self {
StatKey::NodesCreated => "nodes created",
StatKey::RelationshipsCreated => "relationships created",
}
}
/// Human name, used in the error messages.
fn what(self) -> &'static str {
match self {
StatKey::NodesCreated => "nodes",
StatKey::RelationshipsCreated => "relationships",
}
}
}
/// True iff `s` looks like a line from a `GRAPH.QUERY` statistics block.
///
/// The discriminator matters: FalkorDB **omits** a `Relationships created` line
/// when the count is zero, so "line absent" and "reply we cannot read" look
/// identical unless we can prove a stats block was there at all. Every
/// `GRAPH.QUERY` reply carries at least the execution-time line, and every stats
/// line has the shape `<name>: <number>[ unit]`, so either shape counts.
fn looks_like_stats_line(s: &str) -> bool {
let low = s.to_ascii_lowercase();
if low.contains("execution time") || low.contains("cached execution") {
return true;
}
match low.split_once(':') {
Some((_, tail)) => tail
.split_whitespace()
.next()
.is_some_and(|n| n.parse::<f64>().is_ok()),
None => false,
}
}
/// The only place in this crate that can mint a "the server created N" number.
///
/// `Created`'s field is private to this module and the module exports no other
/// constructor, so **outside `server_reported` there is no expression of type
/// `Created` that is not the return value of [`created_from_reply`]**. That is
/// the point, and it is deliberately a type and not a convention: the defect
/// this module exists to prevent was `Ok(rows_sent)` sitting where a
/// server-reported count belonged, and it survived because the unit tests called
/// the parser DIRECTLY and so could not see which value the production call
/// sites actually returned. Both of those call sites were reverted to rows-sent
/// on 2026-08-01 and all 49 tests still passed. With the count behind this type
/// the same revert is a compile error, which no test can be blind to.
mod server_reported {
use super::{FalkorError, Result, StatKey, looks_like_stats_line};
/// A count of entities the SERVER reported creating, in its reply statistics.
///
/// Deliberately NOT `usize`: a bare integer is interchangeable with the rows
/// the client sent, and that interchangeability was the bug.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct Created(usize);
impl Created {
/// Unwrap to a plain count — for summing across windows and for the
/// public API surface, once provenance has been established by type.
pub fn get(self) -> usize {
self.0
}
}
/// Read `key`'s count out of a `GRAPH.QUERY` reply's statistics block.
///
/// The ONE place a "how much did the server actually write" number is
/// produced; the serial flush and the parallel dispatch both route through it
/// (so the two cannot drift, and so a fix lands once). Walks the reply
/// because the stats array's position varies with whether the query returned
/// a result set.
///
/// Three outcomes, all distinguishable:
/// * the line is present → its count.
/// * a stats block is present but the line is not → `0` (FalkorDB omits
/// zeros — measured against a live FalkorDB on 2026-08-01: a
/// `MATCH … CREATE` that matched no endpoint replies with `Cached
/// execution: 0` + the timing line and NO `Relationships created` line at
/// all, and the query SUCCEEDS).
/// * no stats block at all → [`FalkorError::UnreadableStats`], never a
/// silent 0 and never a fallback to rows-sent.
pub fn created_from_reply(reply: &redis::Value, key: StatKey) -> Result<Created> {
fn walk(v: &redis::Value, needle: &str, found: &mut Option<usize>, saw_stats: &mut bool) {
match v {
redis::Value::BulkString(bytes) => {
if let Ok(s) = std::str::from_utf8(bytes) {
scan(s, needle, found, saw_stats);
}
}
redis::Value::SimpleString(s) => scan(s, needle, found, saw_stats),
redis::Value::Array(items) | redis::Value::Set(items) => {
for it in items {
walk(it, needle, found, saw_stats);
}
}
redis::Value::Map(pairs) => {
for (k, val) in pairs {
walk(k, needle, found, saw_stats);
walk(val, needle, found, saw_stats);
}
}
_ => {}
}
}
fn scan(s: &str, needle: &str, found: &mut Option<usize>, saw_stats: &mut bool) {
if looks_like_stats_line(s) {
*saw_stats = true;
}
let low = s.to_ascii_lowercase();
if low.trim_start().starts_with(needle) {
if let Some((_, tail)) = s.split_once(':') {
if let Ok(n) = tail.trim().split_whitespace().next().unwrap_or("").parse() {
*found = Some(n);
}
}
}
}
let mut found = None;
let mut saw_stats = false;
walk(reply, key.needle(), &mut found, &mut saw_stats);
match (found, saw_stats) {
(Some(n), _) => Ok(Created(n)),
(None, true) => Ok(Created(0)),
(None, false) => Err(FalkorError::UnreadableStats(key.what())),
}
}
}
use server_reported::{Created, created_from_reply};
/// `rows` were sent and `created` came back — the phase-level verdict.
///
/// Shared by the serial and parallel paths so "0 created is not a pass" is
/// stated once (the recurring false-green class) rather than re-derived per
/// call site.
fn verdict(created: usize, rows: usize, key: StatKey) -> Result<usize> {
if rows > 0 && created == 0 {
return Err(FalkorError::NothingCreated {
what: key.what(),
rows,
});
}
Ok(created)
}
/// Bulk-loads Arrow `RecordBatch`es into a FalkorDB graph.
///
/// Vertices are inserted with a `_gar_id` property (the row's 0-based absolute
/// index across all batches) so that edge insertion can locate endpoints by
/// internal GraphAr vertex ID without requiring a user-defined primary key.
pub struct FalkorDbLoader {
conn: redis::aio::MultiplexedConnection,
graph: String,
batch_size: usize,
}
impl FalkorDbLoader {
pub async fn connect(url: &str, graph: impl Into<String>) -> Result<Self> {
Self::connect_with_identity(url, graph, None).await
}
/// [`connect`](Self::connect), optionally presenting a **mutual-TLS client identity**
/// over `rediss://` - what a FalkorDB running `tls-auth-clients yes` requires before it
/// will take a single command. `None` is the plain path, unchanged.
pub async fn connect_with_identity(
url: &str,
graph: impl Into<String>,
tls: Option<crate::FalkorTls>,
) -> Result<Self> {
let client = crate::tls::client(url, tls)?;
let conn = client.get_multiplexed_async_connection().await?;
Ok(Self {
conn,
graph: graph.into(),
batch_size: DEFAULT_BATCH_SIZE,
})
}
pub fn with_batch_size(mut self, n: usize) -> Result<Self> {
if n == 0 {
return Err(FalkorError::InvalidBatchSize);
}
self.batch_size = n;
Ok(self)
}
/// Fail a command that gets no reply within `d` instead of awaiting forever.
///
/// **Opt-in, and deliberately not a default.** A legitimate FalkorDB command
/// can take a long time — `MATCH (n) DETACH DELETE n` over 200 000 nodes
/// measured ~10 s on oden — so a ceiling picked here would start failing real
/// work. Callers that know their own ceiling (a test, a CI arm, a bounded
/// demo load) set one; `connect` leaves `None`, so production behaviour is
/// unchanged byte for byte.
///
/// # Why this exists — STINKY-19 R22
///
/// `knut-popviz`'s `falkor_parity` was recorded on 2026-08-01 sitting **24
/// minutes** without completing: client process at 0:00 CPU, all 66 threads
/// parked, and the server holding a partially received 163 KB `graph.QUERY`
/// (`qbuf=163227 qbuf-free=603 idle=182`) whose remaining bytes never
/// arrived. Nothing failed and nothing reported — it simply never returned,
/// because `get_multiplexed_async_connection()` leaves `response_timeout` at
/// `None` and `redis`'s `Pipeline::send_recv` then awaits its oneshot with no
/// deadline at all.
///
/// This does **not** repair whatever wedges the link. It makes the wedge
/// LOUD: a run that used to eat a machine for 24 minutes and then be killed
/// by hand comes back as an error instead. That is the difference between a
/// defect somebody can act on and a box that merely looks busy — and it is
/// the reason nobody noticed for three weeks.
///
/// Guarded by `tests/robot_response_timeout.rs`, which drives a real socket
/// that accepts and never answers.
pub fn with_response_timeout(mut self, d: std::time::Duration) -> Self {
self.conn.set_response_timeout(d);
self
}
/// Insert single-label vertices. Returns the nodes the server reports
/// **creating** (its `Nodes created` statistic), not the rows sent.
///
/// Each node gets a `_gar_id` property equal to its absolute row index
/// across all batches (0-based), which matches GraphAr's internal vertex ID.
///
/// Convenience over [`insert_vertices_multi`] with a one-label slice — the
/// emitted `CREATE (n:`Label`)` is byte-identical to the historical path.
pub async fn insert_vertices(&mut self, label: &str, batches: &[RecordBatch]) -> Result<usize> {
self.insert_vertices_multi(&[label], batches).await
}
/// Insert vertices carrying **multiple labels** — emits
/// `CREATE (n:`A`:`B`:`C`)` so a node can be, e.g., both `Person` and
/// `Employee`. Returns the nodes the server reports **creating**.
///
/// Each node gets a `_gar_id` property equal to its absolute row index
/// across all batches (0-based). A single-element `labels` slice yields the
/// same query as [`insert_vertices`].
pub async fn insert_vertices_multi(
&mut self,
labels: &[&str],
batches: &[RecordBatch],
) -> Result<usize> {
let result = self.insert_vertices_multi_inner(labels, batches).await;
// Emit the UNWIND vertex-load surface's health into the nornir matrix.
crate::functional_status(
"graphar-falkordb/loader",
"insert_vertices",
result.is_ok(),
&match &result {
Ok(n) => format!(
"{n} nodes CREATED as :{} (server statistics)",
labels.join(":")
),
Err(e) => e.to_string(),
},
);
result
}
async fn insert_vertices_multi_inner(
&mut self,
labels: &[&str],
batches: &[RecordBatch],
) -> Result<usize> {
if batches.is_empty() {
return Err(FalkorError::EmptyBatches);
}
if labels.is_empty() {
return Err(FalkorError::ColumnNotFound("at least one label".into()));
}
// Pre-render the (possibly multi-) label suffix once.
let label_suffix = render_labels(labels);
let mut total = 0usize;
let mut created = 0usize;
let mut pending = 0usize;
// One reusable, pre-sized buffer holding the comma-joined row maps for
// the current flush window — no per-row/per-cell String allocations.
let mut maps = String::with_capacity(self.batch_size * 64);
for batch in batches {
// Downcast each column once for the whole batch.
let cols = resolve_cols(batch);
for row in 0..batch.num_rows() {
// The row shape lives in ONE place, shared with the parallel path.
push_vertex_row(&mut maps, &cols, row, total, pending)?;
total += 1;
pending += 1;
if pending >= self.batch_size {
created += self.flush_vertices(&label_suffix, &mut maps).await?.get();
pending = 0;
}
}
}
if pending > 0 {
created += self.flush_vertices(&label_suffix, &mut maps).await?.get();
}
verdict(created, total, StatKey::NodesCreated)
}
/// Flush the accumulated row-maps buffer as one `UNWIND … CREATE` query.
/// `label_suffix` is the already-rendered (single- or multi-) label, e.g.
/// `` `Person` `` or `` `Person`:`Employee` ``.
///
/// Returns the nodes the **server reports creating**, read off the reply's
/// statistics — not the rows we sent.
async fn flush_vertices(&mut self, label_suffix: &str, maps: &mut String) -> Result<Created> {
// Shared with the parallel path (`render_vertex_queries`) so the two
// emit byte-identical Cypher — only the dispatch differs.
let query = build_vertex_query(label_suffix, maps);
let reply = self.run_query(&query).await?;
maps.clear();
created_from_reply(&reply, StatKey::NodesCreated)
}
/// Insert edges from a combined adj-list + (optional) property RecordBatch.
///
/// The batch **must** contain `src` and `dst` columns (Int64, the GraphAr
/// internal vertex IDs). Any additional columns are treated as edge properties.
///
/// Endpoints are matched by `_gar_id` (inserted during `insert_vertices`).
/// Convenience over [`insert_edges_keyed`] with the single `_gar_id` key on
/// both endpoints — the emitted Cypher is identical to the historical path.
pub async fn insert_edges(
&mut self,
src_label: &str,
edge_label: &str,
dst_label: &str,
batches: &[RecordBatch],
) -> Result<usize> {
self.insert_edges_keyed(
src_label,
edge_label,
dst_label,
&["_gar_id"],
&["_gar_id"],
batches,
)
.await
}
/// Insert edges matching endpoints on a **composite key** — one or more
/// properties per endpoint (e.g. `MATCH (a:`L` {`k1`: …, `k2`: …})`).
///
/// `src_keys` / `dst_keys` name the endpoint key *properties*; the batch's
/// carrier columns generalize from `src`/`dst` to **one column per key**:
/// for `src_keys = ["country", "code"]` the batch must carry columns
/// `country` and `code` (and likewise the `dst_keys` columns). Any column
/// not named by a key is an edge property. Key columns may be any
/// Cypher-renderable Arrow type, not only Int64.
///
/// The single-key `["_gar_id"]`/`["_gar_id"]` case keeps `src`/`dst` as the
/// carrier column names, so [`insert_edges`] routes through here unchanged.
pub async fn insert_edges_keyed(
&mut self,
src_label: &str,
edge_label: &str,
dst_label: &str,
src_keys: &[&str],
dst_keys: &[&str],
batches: &[RecordBatch],
) -> Result<usize> {
let result = self
.insert_edges_keyed_inner(
src_label, edge_label, dst_label, src_keys, dst_keys, batches,
)
.await;
// Emit the UNWIND edge-load surface's health into the nornir matrix.
crate::functional_status(
"graphar-falkordb/loader",
"insert_edges",
result.is_ok(),
&match &result {
Ok(n) => format!(
"{n} :{edge_label} relationships CREATED ({src_label}->{dst_label}, server statistics)"
),
Err(e) => e.to_string(),
},
);
result
}
#[allow(clippy::too_many_arguments)]
async fn insert_edges_keyed_inner(
&mut self,
src_label: &str,
edge_label: &str,
dst_label: &str,
src_keys: &[&str],
dst_keys: &[&str],
batches: &[RecordBatch],
) -> Result<usize> {
if batches.is_empty() {
return Err(FalkorError::EmptyBatches);
}
if src_keys.is_empty() || dst_keys.is_empty() {
return Err(FalkorError::ColumnNotFound(
"at least one key per endpoint".into(),
));
}
// The edge lane knows its own endpoint keys, so IT guarantees the
// index — `MATCH (a:L {k: row._src})` without one plans a label scan
// per endpoint, O(nodes) per row (measured: the ~2.7k edges/s lane).
{
let mut conn = self.conn.clone();
ensure_edge_indexes(&mut conn, &self.graph, src_label, src_keys, dst_label, dst_keys)
.await?;
}
// For the single legacy `_gar_id` key the carrier columns are `src`/`dst`;
// for composite keys they are the key property names themselves.
let single_legacy = src_keys == ["_gar_id"] && dst_keys == ["_gar_id"];
let src_cols_wanted: Vec<&str> = if single_legacy {
vec!["src"]
} else {
src_keys.to_vec()
};
let dst_cols_wanted: Vec<&str> = if single_legacy {
vec!["dst"]
} else {
dst_keys.to_vec()
};
let mut total = 0usize;
let mut created = 0usize;
let mut row_buf: Vec<EdgeRow> = Vec::with_capacity(self.batch_size);
for batch in batches {
let schema = batch.schema();
// Resolve the (composite) key carrier columns in declared order.
let src_key_arrs: Vec<&dyn Array> = src_cols_wanted
.iter()
.map(|name| {
batch
.column_by_name(name)
.map(|c| c.as_ref())
.ok_or_else(|| FalkorError::ColumnNotFound((*name).into()))
})
.collect::<Result<_>>()?;
let dst_key_arrs: Vec<&dyn Array> = dst_cols_wanted
.iter()
.map(|name| {
batch
.column_by_name(name)
.map(|c| c.as_ref())
.ok_or_else(|| FalkorError::ColumnNotFound((*name).into()))
})
.collect::<Result<_>>()?;
// Prop columns = everything that isn't a key carrier column.
let prop_cols: Vec<(&str, &dyn Array)> = schema
.fields()
.iter()
.zip(batch.columns())
.filter(|(f, _)| {
let n = f.name().as_str();
!src_cols_wanted.contains(&n) && !dst_cols_wanted.contains(&n)
})
.map(|(f, col)| (f.name().as_str(), col.as_ref()))
.collect();
for row in 0..batch.num_rows() {
let src_id: Vec<String> = src_key_arrs
.iter()
.map(|arr| arrow_value_to_cypher(*arr, row))
.collect::<Result<_>>()?;
let dst_id: Vec<String> = dst_key_arrs
.iter()
.map(|arr| arrow_value_to_cypher(*arr, row))
.collect::<Result<_>>()?;
let props: Vec<(String, String)> = prop_cols
.iter()
.map(|(name, arr)| {
let val = arrow_value_to_cypher(*arr, row)?;
Ok((name.to_string(), val))
})
.collect::<Result<_>>()?;
row_buf.push(EdgeRow {
src_id,
dst_id,
props,
});
total += 1;
if row_buf.len() >= self.batch_size {
created += self
.flush_edges(
src_label,
edge_label,
dst_label,
src_keys,
dst_keys,
&mut row_buf,
)
.await?
.get();
}
}
}
if !row_buf.is_empty() {
created += self
.flush_edges(
src_label,
edge_label,
dst_label,
src_keys,
dst_keys,
&mut row_buf,
)
.await?
.get();
}
verdict(created, total, StatKey::RelationshipsCreated)
}
async fn flush_edges(
&mut self,
src_label: &str,
edge_label: &str,
dst_label: &str,
src_keys: &[&str],
dst_keys: &[&str],
buf: &mut Vec<EdgeRow>,
) -> Result<Created> {
// Parameterized window: the rows ride in the `CYPHER rows=` prefix, so
// the query body is one constant string per (labels × keys) shape and
// FalkorDB's plan cache serves every window after the first.
let query =
unwind_edge_query_keyed_param(src_label, edge_label, dst_label, src_keys, dst_keys, buf);
let reply = self.run_query(&query).await?;
buf.clear();
// The relationships the SERVER created. This query is `MATCH … CREATE`:
// it succeeds whether or not the endpoints matched, so the reply is the
// only witness that anything was written.
created_from_reply(&reply, StatKey::RelationshipsCreated)
}
/// Execute a raw Cypher query against the configured graph.
pub async fn run_query(&mut self, query: &str) -> Result<redis::Value> {
graph_query(&mut self.conn, &self.graph, query).await
}
/// Create an index on a property to speed up edge MATCH lookups —
/// **idempotent**: a no-op when the index already exists.
///
/// FalkorDB's bare `CREATE INDEX FOR (n:L) ON (n.p)` errors with
/// `Attribute: '<p>' is already indexed` when the index is already present
/// (a re-used or twice-populated graph). In the population pipeline that
/// error would ABORT the load *before the edge phase* — leaving a graph with
/// nodes but **zero edges** (the live `korp_demo` regression: 121200 nodes,
/// 0 edges). This makes a redundant create a success while still surfacing
/// every *other* error.
///
/// Strategy — portable across FalkorDB versions (no version sniffing):
/// 1. If `CALL db.indexes()` already lists `(label, property)`, skip the
/// create entirely (the common warm-graph path).
/// 2. Otherwise issue the `CREATE INDEX` and treat **only** the
/// "already indexed" server error as success (covers the race where a
/// concurrent create won, and any version whose `db.indexes()` shape we
/// don't recognise). Any non-"already indexed" error still propagates.
pub async fn create_index(&mut self, label: &str, property: &str) -> Result<()> {
ensure_index(&mut self.conn, &self.graph, label, property).await
}
// -----------------------------------------------------------------------
// Parallel bulk path — the fix for the ~1-core FalkorDB ingest stall.
//
// The serial `insert_vertices`/`insert_edges_keyed` await one UNWIND-CREATE
// query at a time over the single link, so the client sits idle for a full
// round-trip between queries — the ingest is stuck at ~1 core no matter how
// many the box has. `redis::aio::MultiplexedConnection` is `Clone` and
// carries many concurrent in-flight commands over the one link, so the
// parallel path renders the SAME per-window queries (byte-identical Cypher,
// same `_gar_id` scheme — see `render_vertex_queries`) and dispatches them
// with up to `concurrency` outstanding via `buffer_unordered`, saturating the
// pipeline. `&self` (not `&mut self`): each in-flight task owns a cloned
// connection. Additive — the serial methods are untouched.
// -----------------------------------------------------------------------
/// Concurrent single-label vertex bulk-insert — the parallel sibling of
/// [`insert_vertices`](Self::insert_vertices). Renders the identical
/// `UNWIND … CREATE` queries and dispatches up to `concurrency` at once.
/// Because vertices are `CREATE`d over disjoint `_gar_id` ranges the result
/// is order-independent, so concurrency cannot change the produced graph.
/// Returns the nodes the server reports **creating**.
pub async fn insert_vertices_par(
&self,
label: &str,
batches: &[RecordBatch],
concurrency: usize,
) -> Result<usize> {
self.insert_vertices_multi_par(&[label], batches, concurrency, |_| {})
.await
}
/// Concurrent multi-label vertex bulk-insert with a per-completed-query
/// `on_progress(rows)` callback (fired as each window lands, so a live
/// driver can advance progress). Emits the `insert_vertices_par` matrix cell.
pub async fn insert_vertices_multi_par(
&self,
labels: &[&str],
batches: &[RecordBatch],
concurrency: usize,
on_progress: impl FnMut(usize),
) -> Result<usize> {
let result = self
.insert_vertices_multi_par_inner(labels, batches, concurrency, on_progress)
.await;
crate::functional_status(
"graphar-falkordb/loader",
"insert_vertices_par",
result.is_ok(),
&match &result {
Ok(n) => format!(
"{n} nodes CREATED as :{} (concurrent x{concurrency}, server statistics)",
labels.join(":")
),
Err(e) => e.to_string(),
},
);
result
}
async fn insert_vertices_multi_par_inner(
&self,
labels: &[&str],
batches: &[RecordBatch],
concurrency: usize,
mut on_progress: impl FnMut(usize),
) -> Result<usize> {
let (queries, total) = render_vertex_queries(labels, batches, self.batch_size)?;
let created = self
.dispatch_par(
queries.into_iter().map(|(q, _rows)| q).collect(),
concurrency,
&mut on_progress,
StatKey::NodesCreated,
)
.await?;
verdict(created, total, StatKey::NodesCreated)
}
/// Concurrent edge bulk-insert — the parallel sibling of
/// [`insert_edges`](Self::insert_edges) (legacy `_gar_id` endpoints).
pub async fn insert_edges_par(
&self,
src_label: &str,
edge_label: &str,
dst_label: &str,
batches: &[RecordBatch],
concurrency: usize,
) -> Result<usize> {
self.insert_edges_keyed_par(
src_label,
edge_label,
dst_label,
&["_gar_id"],
&["_gar_id"],
batches,
concurrency,
|_| {},
)
.await
}
/// Concurrent composite-key edge bulk-insert with a per-completed-query
/// `on_progress(rows)` callback — the parallel sibling of
/// [`insert_edges_keyed`](Self::insert_edges_keyed). Runs AFTER all vertices
/// exist (the caller's barrier); each edge query `MATCH`es existing endpoints
/// then `CREATE`s a fresh relationship, so concurrent execution is
/// order-independent and cannot change the produced graph.
#[allow(clippy::too_many_arguments)]
pub async fn insert_edges_keyed_par(
&self,
src_label: &str,
edge_label: &str,
dst_label: &str,
src_keys: &[&str],
dst_keys: &[&str],
batches: &[RecordBatch],
concurrency: usize,
on_progress: impl FnMut(usize),
) -> Result<usize> {
let result = self
.insert_edges_keyed_par_inner(
src_label,
edge_label,
dst_label,
src_keys,
dst_keys,
batches,
concurrency,
on_progress,
)
.await;
crate::functional_status(
"graphar-falkordb/loader",
"insert_edges_par",
result.is_ok(),
&match &result {
Ok(n) => format!(
"{n} :{edge_label} relationships CREATED ({src_label}->{dst_label}, concurrent x{concurrency}, server statistics)"
),
Err(e) => e.to_string(),
},
);
result
}
#[allow(clippy::too_many_arguments)]
async fn insert_edges_keyed_par_inner(
&self,
src_label: &str,
edge_label: &str,
dst_label: &str,
src_keys: &[&str],
dst_keys: &[&str],
batches: &[RecordBatch],
concurrency: usize,
mut on_progress: impl FnMut(usize),
) -> Result<usize> {
// Same endpoint-index guarantee as the serial lane, on a cloned
// connection (this path is `&self`).
{
let mut conn = self.conn.clone();
ensure_edge_indexes(&mut conn, &self.graph, src_label, src_keys, dst_label, dst_keys)
.await?;
}
let (queries, total) = render_edge_queries(
src_label,
edge_label,
dst_label,
src_keys,
dst_keys,
batches,
self.batch_size,
)?;
let created = self
.dispatch_par(
queries.into_iter().map(|(q, _rows)| q).collect(),
concurrency,
&mut on_progress,
StatKey::RelationshipsCreated,
)
.await?;
verdict(created, total, StatKey::RelationshipsCreated)
}
/// Execute pre-rendered `(query, rows)` windows with up to `concurrency`
/// in flight over CLONED connections, firing `on_progress(created)` as each
/// completes and returning the total the server reports creating. The single
/// point where the serial one-RTT-at-a-time link becomes a saturated
/// pipeline. Any query error aborts (returned).
///
/// `on_progress` reports **created** entities, not rows dispatched. That is
/// the difference between a progress meter and a wish: a live driver that
/// advances on rows sent fills to 100 % against a graph the server wrote
/// nothing into.
///
/// The scheduling itself lives in the connection-free, generic
/// [`buffered_dispatch`] (unit-testable that it truly runs `concurrency`
/// tasks at once); here we only supply the per-window work: clone the
/// (`Clone`) multiplexed connection, run one `GRAPH.QUERY` and read its
/// statistics via the shared [`created_from_reply`].
/// Takes **`Vec<String>` and not `Vec<(String, rows)>` on purpose.** The
/// per-window row count is not needed here, and while it was in scope the
/// closure could — and originally did — return it instead of the server's
/// count. Not passing it in is a cheaper guarantee than any test: there is no
/// rows value in this function to return by mistake.
async fn dispatch_par(
&self,
queries: Vec<String>,
concurrency: usize,
on_progress: &mut impl FnMut(usize),
key: StatKey,
) -> Result<usize> {
let graph = self.graph.clone();
let base = self.conn.clone();
buffered_dispatch(
queries,
concurrency,
|q| {
let mut conn = base.clone();
let graph = graph.clone();
async move {
let reply = graph_query(&mut conn, &graph, &q).await?;
created_from_reply(&reply, key).map(Created::get)
}
},
on_progress,
)
.await
}
// -----------------------------------------------------------------------
// Native GRAPH.BULK fast-lane — the deeper lever below the parallel-UNWIND
// pipeline. Instead of `UNWIND … CREATE` (the server parses + plans +
// executes per batch) it ships FalkorDB's binary, typed, columnar
// bulk-insert payload, which the server ingests WITHOUT the Cypher planner —
// the initial/empty-graph population fast path. The encoder is pure Rust in
// `crate::bulk` (no CSV, no Python, no shell — L9), fed straight from the
// same Arrow batches. Additive: `UNWIND` stays the incremental/upsert path.
// -----------------------------------------------------------------------
/// Native `GRAPH.BULK` initial population — encode `nodes`/`edges` and ship
/// the whole graph in **one** `GRAPH.BULK` round trip. It `CREATE`s (never
/// MERGEs), so it is for a fresh/empty key (`--drop`-then-load), chosen by
/// [`choose_bulk`] on an empty graph; the `UNWIND` path remains the
/// incremental/MERGE fallback. Returns
/// `(nodes, relations)` loaded. Each node carries the same `_gar_id` property
/// `insert_vertices` writes, so the produced graph is byte-identical (the
/// container-gated parity test proves it).
pub async fn bulk_insert(
&mut self,
nodes: &[BulkNodeGroup<'_>],
edges: &[BulkEdgeGroup<'_>],
) -> Result<(usize, usize)> {
let result = self.bulk_insert_inner(nodes, edges, false).await;
crate::functional_status(
"graphar-falkordb/loader",
"bulk_insert",
result.is_ok(),
&match &result {
Ok((n, e)) => format!("{n} nodes + {e} relations via GRAPH.BULK"),
Err(e) => e.to_string(),
},
);
result
}
/// Streamed/appendable `GRAPH.BULK` — send each node group, then each edge
/// group, as its **own** appendable `GRAPH.BULK` call (`BEGIN` on the first)
/// instead of one giant command. Bounds per-command size/memory for very
/// large loads and is the seam for parallel-graph sharding. Ordered on
/// purpose: the server assigns global node ids in call order, and every node
/// must exist before its edges. Byte-identical result to [`bulk_insert`].
/// Streamed `GRAPH.BULK` of **nodes only, verbatim**: each node carries
/// exactly its batch's columns, no synthesized `_gar_id` (see
/// [`crate::bulk::build_bulk_payload_verbatim`]). Appends to an existing
/// graph key, or creates it. `CREATE`, never `MERGE`: the caller guarantees
/// the labels are empty and the keys unique. Returns nodes created.
pub async fn bulk_insert_nodes_verbatim(&mut self, nodes: &[BulkNodeGroup<'_>]) -> Result<usize> {
let result = self.bulk_insert_inner_with_progress(nodes, &[], true, PayloadKind::VerbatimNodes, |_, _| {}).await;
crate::functional_status(
"graphar-falkordb/loader",
"bulk_insert_nodes_verbatim",
result.is_ok(),
&match &result {
Ok((n, _)) => format!("{n} nodes via streamed GRAPH.BULK (verbatim)"),
Err(e) => e.to_string(),
},
);
Ok(result?.0)
}
/// Streamed `GRAPH.BULK` of **relationships only**, between nodes already in
/// the graph: `src`/`dst` are absolute node ids (see
/// [`crate::bulk::build_bulk_payload_edges_absolute`]). Appends to the existing
/// graph key. `CREATE`, never `MERGE`: the caller guarantees the relationship
/// type is empty and the rows unique. Returns relationships created.
pub async fn bulk_insert_edges_absolute(&mut self, edges: &[BulkEdgeGroup<'_>]) -> Result<usize> {
let result = self.bulk_insert_inner_with_progress(&[], edges, true, PayloadKind::AbsoluteEdges, |_, _| {}).await;
crate::functional_status(
"graphar-falkordb/loader",
"bulk_insert_edges_absolute",
result.is_ok(),
&match &result {
Ok((_, r)) => format!("{r} relations via streamed GRAPH.BULK (absolute ids)"),
Err(e) => e.to_string(),
},
);
Ok(result?.1)
}
pub async fn bulk_insert_streamed(
&mut self,
nodes: &[BulkNodeGroup<'_>],
edges: &[BulkEdgeGroup<'_>],
) -> Result<(usize, usize)> {
self.bulk_insert_streamed_with_progress(nodes, edges, |_, _| {})
.await
}
/// [`bulk_insert_streamed`](Self::bulk_insert_streamed) that **reports progress
/// as it streams**. `on_progress` is called after each blob lands with the
/// *cumulative* `(nodes, relations)` written so far.
///
/// Why this exists: the streamed lane already sends one appendable `GRAPH.BULK`
/// per blob, but it reported nothing until the whole population had landed — so a
/// UI driving a progress bar off it had exactly two observable states, 0% and
/// 100%. Since blobs are 1:1 with the caller's node/edge groups and are sent in
/// order, the cumulative count after each blob is *exact* — this is real measured
/// progress, not an interpolation.
pub async fn bulk_insert_streamed_with_progress(
&mut self,
nodes: &[BulkNodeGroup<'_>],
edges: &[BulkEdgeGroup<'_>],
on_progress: impl FnMut(usize, usize),
) -> Result<(usize, usize)> {
let result = self
.bulk_insert_inner_with_progress(nodes, edges, true, PayloadKind::Standard, on_progress)
.await;
crate::functional_status(
"graphar-falkordb/loader",
"bulk_insert_streamed",
result.is_ok(),
&match &result {
Ok((n, e)) => format!("{n} nodes + {e} relations via streamed GRAPH.BULK"),
Err(e) => e.to_string(),
},
);
result
}
async fn bulk_insert_inner(
&mut self,
nodes: &[BulkNodeGroup<'_>],
edges: &[BulkEdgeGroup<'_>],
streamed: bool,
) -> Result<(usize, usize)> {
self.bulk_insert_inner_with_progress(nodes, edges, streamed, PayloadKind::Standard, |_, _| {})
.await
}
async fn bulk_insert_inner_with_progress(
&mut self,
nodes: &[BulkNodeGroup<'_>],
edges: &[BulkEdgeGroup<'_>],
streamed: bool,
kind: PayloadKind,
mut on_progress: impl FnMut(usize, usize),
) -> Result<(usize, usize)> {
// The blob encode fans across cores (gatling); time it and report how many
// cores it could use (== #blobs capped at available parallelism) so the log
// shows the fill is multicore, not the old one-core-per-label encode. One
// blob is one `GRAPH.BULK` command bounded under `proto-max-bulk-len`.
let enc_t0 = std::time::Instant::now();
let payload = match kind {
PayloadKind::Standard => build_bulk_payload(nodes, edges)?,
PayloadKind::VerbatimNodes => crate::bulk::build_bulk_payload_verbatim(nodes)?,
PayloadKind::AbsoluteEdges => crate::bulk::build_bulk_payload_edges_absolute(edges)?,
};
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
let blobs = payload.blobs.len();
eprintln!(
"graphar-falkordb bulk: encoded {blobs} GRAPH.BULK blob(s) across up to {} core(s) \
in {} ms ({} nodes + {} relations; each blob < proto-max-bulk-len)",
cores.min(blobs.max(1)),
enc_t0.elapsed().as_millis(),
payload.node_count,
payload.relation_count,
);
let totals = (payload.node_count as usize, payload.relation_count as usize);
// `BEGIN` creates the graph and errors if the key already exists;
// tokenless append errors if the key is absent. So the FIRST call needs
// `BEGIN` iff the key does not yet exist (a truly fresh graph); a graph
// the auto-select probe already materialized appends without it.
let mut begin = !self.key_exists().await;
if streamed {
// One appendable GRAPH.BULK per blob, in node-then-edge order; the
// server keeps assigning global node ids across the ordered calls.
// Cumulative counts reported to `on_progress` after each blob lands —
// exact, because blobs are 1:1 with the caller's groups and ordered.
let (mut nodes_done, mut rels_done) = (0usize, 0usize);
for (i, blob) in payload.blobs.iter().enumerate() {
let is_node = i < payload.node_blob_len;
let (nc, rc, node_blobs) = if is_node {
(payload.blob_counts[i], 0, 1)
} else {
(0, payload.blob_counts[i], 0)
};
bulk_command(
&mut self.conn,
&self.graph,
begin,
nc,
rc,
node_blobs,
&[blob.as_slice()],
)
.await?;
begin = false; // only the first call creates the graph
nodes_done += nc as usize;
rels_done += rc as usize;
on_progress(nodes_done, rels_done);
}
} else {
let refs: Vec<&[u8]> = payload.blobs.iter().map(|b| b.as_slice()).collect();
bulk_command(
&mut self.conn,
&self.graph,
begin,
payload.node_count,
payload.relation_count,
payload.node_blob_len,
&refs,
)
.await?;
// The single-command lane has no intermediate point to report from.
on_progress(totals.0, totals.1);
}
Ok(totals)
}
/// Whether the graph's Redis key currently exists (a non-mutating `EXISTS`,
/// so — unlike a `GRAPH.QUERY` probe — it does not itself materialize the
/// key). Drives the `BEGIN`-vs-append choice in [`Self::bulk_insert`]. Any
/// error counts as "absent" so the first bulk call takes the creating path.
async fn key_exists(&mut self) -> bool {
redis::cmd("EXISTS")
.arg(&self.graph)
.query_async::<i64>(&mut self.conn)
.await
.map(|n| n > 0)
.unwrap_or(false)
}
/// Whether the configured graph currently holds **zero** nodes — the signal
/// [`choose_bulk`] uses to pick the `GRAPH.BULK` fast path (an empty key is a
/// fresh initial population; a non-empty one needs the incremental `UNWIND`
/// path). A missing graph key, or any error reading it, counts as empty
/// (the load is about to create it).
pub async fn graph_is_empty(&mut self) -> bool {
match self.run_query("MATCH (n) RETURN count(n)").await {
Ok(reply) => first_int(&reply) == Some(0),
Err(_) => true,
}
}
/// Whether the graph currently holds **zero** nodes of `label` — the
/// per-label analogue of [`graph_is_empty`], for a caller (e.g. the bifrost
/// `FalkorDbSink`) that lands one label at a time and wants the `GRAPH.BULK`
/// fast lane whenever *that* label is fresh, even if other labels already
/// exist in the same graph.
///
/// `GRAPH.BULK` `CREATE`s (it cannot MERGE), so it is sound only when the
/// label has no live nodes to duplicate. The two branches encode that:
/// * **No graph key yet** — a truly fresh load; every label is empty and (as
/// in `graph_is_empty`) we want the fast lane, so return `true`.
/// * **Key exists** — only an authoritative `count(n) == 0` is safe to bulk
/// onto. An ambiguous read *error* is deliberately treated as **non-empty**
/// (`false`) so the caller falls back to the idempotent `UNWIND`/MERGE path
/// rather than risk `CREATE`-duplicating a live node. (This is the opposite
/// default to `graph_is_empty`, whose error-is-empty is fine only because
/// its sole caller, popviz, gates a whole-graph one-shot load.)
pub async fn label_is_empty(&mut self, label: &str) -> bool {
if !self.key_exists().await {
return true;
}
match self
.run_query(&format!("MATCH (n:`{label}`) RETURN count(n)"))
.await
{
Ok(reply) => first_int(&reply) == Some(0),
Err(_) => false,
}
}
}
/// Which `GRAPH.BULK` payload a bulk load encodes.
#[derive(Clone, Copy, Debug)]
enum PayloadKind {
/// Node + edge groups, a synthesized `_gar_id`, edges by `_gar_id` position.
Standard,
/// Nodes only, exactly their columns.
VerbatimNodes,
/// Edges only, by absolute node id.
AbsoluteEdges,
}
/// Auto-select policy for initial vs incremental population: the native
/// `GRAPH.BULK` fast lane is chosen **only** for an empty graph (it `CREATE`s
/// and cannot MERGE, so it is safe only on a fresh key); a non-empty graph falls
/// back to the parallel-`UNWIND` path. Pure so the decision is unit-tested
/// without a DB (RED-when-broken: flip it and bulk would clobber a live graph).
pub fn choose_bulk(graph_is_empty: bool) -> bool {
graph_is_empty
}
/// First integer anywhere in a GRAPH.QUERY reply (the `count(n)` scalar).
fn first_int(v: &redis::Value) -> Option<i64> {
match v {
redis::Value::Int(i) => Some(*i),
redis::Value::Array(items) | redis::Value::Set(items) => items.iter().find_map(first_int),
redis::Value::BulkString(b) => String::from_utf8_lossy(b).trim().parse().ok(),
_ => None,
}
}
/// Send one
/// `GRAPH.BULK <graph> [BEGIN] <node_count> <relation_count> <label_blob_count> <reltype_blob_count> <node_blob…> <reltype_blob…>`
/// — the FalkorDB bulk wire: the optional `BEGIN` create-token, the two entity
/// counts, then the count of node (label) blobs and of relation (reltype) blobs,
/// then the blobs in that order.
///
/// `begin` = `true` **creates** the graph (first call to an absent key) and
/// errors if the key already exists; `false` **appends** and errors if the key
/// is absent (both verified live) — [`FalkorDbLoader::bulk_insert_inner`] picks
/// it from `EXISTS`. `blobs` is `node_blob_count` node blobs then the reltype
/// blobs; all binary RESP bulk-string args. Shared by the single-shot and
/// streamed senders so both hit the wire identically.
#[allow(clippy::too_many_arguments)]
async fn bulk_command(
conn: &mut redis::aio::MultiplexedConnection,
graph: &str,
begin: bool,
node_count: u64,
relation_count: u64,
node_blob_count: usize,
blobs: &[&[u8]],
) -> Result<()> {
let reltype_blob_count = blobs.len() - node_blob_count;
let mut cmd = redis::cmd("GRAPH.BULK");
cmd.arg(graph);
if begin {
cmd.arg("BEGIN");
}
cmd.arg(node_count)
.arg(relation_count)
.arg(node_blob_count)
.arg(reltype_blob_count);
for blob in blobs {
cmd.arg(*blob);
}
let _: redis::Value = cmd.query_async(conn).await?;
Ok(())
}
/// Run `items` through `run_one` with up to `concurrency` futures **in flight
/// at once** ([`znippy_zoomies::gatling::io::run`] — LAW #0: concurrency goes
/// through gatling, the same async engine the doris sink rides), firing
/// `on_progress(rows)` as each completes; any error aborts (returned). This is
/// the connection-free heart of the parallel bulk path — the single place the
/// serial one-RTT-at-a-time link becomes a saturated pipeline. It is generic
/// over the per-item work precisely so the concurrency contract can be proven
/// WITHOUT a live DB: a fake `run_one` that records the peak simultaneous
/// in-flight count asserts the engine really overlaps up to `concurrency` tasks
/// (RED if this ever regressed to a serial `for … await` — max in-flight would
/// drop to 1). `concurrency` is floored at 1 (by the engine too).
///
/// **Unordered completion is deliberate**, not an accident of the old
/// `buffer_unordered`: the UNWIND windows are independent MERGE/CREATE queries
/// and the only consumed outputs are the order-insensitive created-count sum
/// and the progress ticks — re-sequencing (`gatling::io::run_ordered`, the
/// doris sink's choice, which must label chunks in submission order) would only
/// delay progress behind the slowest earlier window for nothing.
pub async fn buffered_dispatch<T, Fut, F>(
items: Vec<T>,
concurrency: usize,
run_one: F,
mut on_progress: impl FnMut(usize),
) -> Result<usize>
where
F: Fn(T) -> Fut,
Fut: std::future::Future<Output = Result<usize>>,
{
let mut sum = 0usize;
let jobs = items.into_iter().map(|it| {
let fut = run_one(it);
async move { fut.await.map_err(anyhow::Error::from) }
});
match znippy_zoomies::gatling::io::run(jobs, concurrency, |n| {
sum += n;
on_progress(n);
Ok(())
})
.await
{
Ok(()) => Ok(sum),
// The only error source is `run_one` (the reducer above is infallible),
// so the anyhow round-trip always downcasts back to the ORIGINAL
// FalkorError value — proven by `dispatch_error_survives_the_engine`.
Err(e) => Err(e
.downcast::<FalkorError>()
.expect("buffered_dispatch jobs only raise FalkorError")),
}
}
/// Run one `GRAPH.QUERY` against `graph` on `conn`. Shared by the serial
/// [`FalkorDbLoader::run_query`] and the parallel [`FalkorDbLoader::dispatch_par`]
/// so both hit the wire exactly the same way.
async fn graph_query(
conn: &mut redis::aio::MultiplexedConnection,
graph: &str,
query: &str,
) -> Result<redis::Value> {
let result: redis::Value = redis::cmd("GRAPH.QUERY")
.arg(graph)
.arg(query)
.query_async(conn)
.await?;
Ok(result)
}
/// Connection-level body of [`FalkorDbLoader::create_index`], shared with the
/// `&self` parallel edge path (which runs it on a cloned multiplexed
/// connection). Same contract: idempotent, only the "already indexed" server
/// error is treated as success.
async fn ensure_index(
conn: &mut redis::aio::MultiplexedConnection,
graph: &str,
label: &str,
property: &str,
) -> Result<()> {
// Best-effort pre-check (see `reply_lists_index`): any failure or unknown
// shape just falls through to the authoritative CREATE below.
if let Ok(reply) = graph_query(conn, graph, "CALL db.indexes()").await
&& reply_lists_index(&reply, label, property)
{
return Ok(());
}
let query = format!("CREATE INDEX FOR (n:`{label}`) ON (n.`{property}`)");
match graph_query(conn, graph, &query).await {
Ok(_) => Ok(()),
// The index we wanted already exists → success. Nothing else swallowed.
Err(e) if is_already_indexed(&e) => Ok(()),
Err(e) => Err(e),
}
}
/// Guarantee every endpoint-key property of an edge load is indexed on its
/// label **before** the first window runs. One `ensure_index` per (label, key
/// property): FalkorDB keeps ONE index per label and coalesces per-property
/// creates into it, and the composite `MATCH (a:L {k1: …, k2: …})` plans a
/// `Node By Index Scan` off that coalesced index (probed live 2026-10-20) — so
/// composite keys need no separate composite-index syntax here.
async fn ensure_edge_indexes(
conn: &mut redis::aio::MultiplexedConnection,
graph: &str,
src_label: &str,
src_keys: &[&str],
dst_label: &str,
dst_keys: &[&str],
) -> Result<()> {
for key in src_keys {
ensure_index(conn, graph, src_label, key).await?;
}
for key in dst_keys {
ensure_index(conn, graph, dst_label, key).await?;
}
Ok(())
}
/// True iff `e` is the FalkorDB "attribute already indexed" server error — the
/// **only** error [`FalkorDbLoader::create_index`] treats as success (the index
/// it wanted already exists). Matches the message case-insensitively so wording
/// drift across versions (`'_gar_id' is already indexed` /
/// `Attribute: '_gar_id' is already indexed`) still counts; every other error
/// (syntax, connection, unrelated attribute failures) returns `false` and
/// propagates. RED-when-broken: swallow every error here and an unrelated
/// failure would be silently ignored — `already_indexed_matches_only_that_error`
/// guards the boundary DB-free.
fn is_already_indexed(e: &FalkorError) -> bool {
matches!(e, FalkorError::Redis(_))
&& e.to_string()
.to_ascii_lowercase()
.contains("already indexed")
}
/// Best-effort scan of a `CALL db.indexes()` reply for an index on
/// `(label, property)`. FalkorDB's index rows carry the label and the indexed
/// property names as strings, nested at version-dependent positions, so rather
/// than parse positionally we check the reply mentions BOTH the label and the
/// property as string cells. This is a pre-check optimization only — a false
/// positive at worst skips a redundant create, a false negative retries it, and
/// [`FalkorDbLoader::create_index`]'s "already indexed" catch is the correctness
/// backstop — so the crude presence test cannot corrupt the load either way.
fn reply_lists_index(reply: &redis::Value, label: &str, property: &str) -> bool {
fn mentions(v: &redis::Value, needle: &str) -> bool {
match v {
redis::Value::BulkString(b) => String::from_utf8_lossy(b) == needle,
redis::Value::SimpleString(s) => s == needle,
redis::Value::Array(items) | redis::Value::Set(items) => {
items.iter().any(|it| mentions(it, needle))
}
_ => false,
}
}
mentions(reply, label) && mentions(reply, property)
}
/// Wrap an already-built row-maps buffer as one `UNWIND … CREATE` vertex query.
/// Shared by the serial flush and the parallel renderer so the emitted Cypher
/// is byte-identical between the two paths.
fn build_vertex_query(label_suffix: &str, maps: &str) -> String {
let mut query = String::with_capacity(maps.len() + label_suffix.len() + 48);
let _ = write!(
query,
"UNWIND [{maps}] AS row CREATE (n:{label_suffix}) SET n = row"
);
query
}
/// Downcast a batch's columns once for the whole batch. Shared so the serial and
/// parallel vertex paths resolve renderers identically.
fn resolve_cols(batch: &RecordBatch) -> Vec<(&str, ColRenderer<'_>)> {
batch
.schema_ref()
.fields()
.iter()
.zip(batch.columns())
.map(|(f, col)| (f.name().as_str(), ColRenderer::resolve(col.as_ref())))
.collect()
}
/// Append ONE vertex row's Cypher map to the flush-window buffer.
///
/// **This is the only place that decides what a vertex row looks like on the
/// wire** — including the `_gar_id` carrier (the absolute 0-based row index
/// across all batches) that `insert_edges` later matches endpoints by.
///
/// It exists because the shape used to be written out twice: once in the serial
/// `insert_vertices_multi_inner` and once in the parallel `render_vertex_queries`,
/// with only doc comments asserting the two agreed. They were never compared, and
/// only the parallel copy had tests — dropping `_gar_id` from the SERIAL call site
/// left all 42 tests green while `insert_vertices` silently stopped emitting the
/// property every edge insert depends on.
///
/// `pending` is how many rows are already in the window (0 ⇒ no separator).
/// `no_second_row_renderer` in `tests` keeps this the only writer.
fn push_vertex_row(
maps: &mut String,
cols: &[(&str, ColRenderer)],
row: usize,
gar_id: usize,
pending: usize,
) -> Result<()> {
if pending > 0 {
maps.push_str(", ");
}
// _gar_id: absolute vertex index across all batches.
let gar = gar_id.to_string();
write_row_map(maps, cols, row, &[("_gar_id", &gar)])
}
/// Render the per-flush-window `UNWIND … CREATE` vertex queries — the pure,
/// connection-free half shared by the serial and parallel vertex loaders so the
/// two emit byte-identical Cypher (only the dispatch differs). Returns each
/// window's `(query, rows_in_window)` plus the grand total. `_gar_id` is the
/// absolute 0-based row index across all batches, exactly as the serial path
/// assigns it — so the produced graph is identical regardless of the order the
/// queries are later executed in (this is what makes the parallel dispatch safe).
fn render_vertex_queries(
labels: &[&str],
batches: &[RecordBatch],
batch_size: usize,
) -> Result<(Vec<(String, usize)>, usize)> {
if batches.is_empty() {
return Err(FalkorError::EmptyBatches);
}
if labels.is_empty() {
return Err(FalkorError::ColumnNotFound("at least one label".into()));
}
let label_suffix = render_labels(labels);
let mut out: Vec<(String, usize)> = Vec::new();
let mut total = 0usize;
let mut pending = 0usize;
let mut maps = String::with_capacity(batch_size * 64);
for batch in batches {
let cols = resolve_cols(batch);
for row in 0..batch.num_rows() {
// Same single row-shape writer the serial path uses.
push_vertex_row(&mut maps, &cols, row, total, pending)?;
total += 1;
pending += 1;
if pending >= batch_size {
out.push((build_vertex_query(&label_suffix, &maps), pending));
maps.clear();
pending = 0;
}
}
}
if pending > 0 {
out.push((build_vertex_query(&label_suffix, &maps), pending));
}
Ok((out, total))
}
/// Render the per-flush-window edge queries — the connection-free half of the
/// parallel edge loader, mirroring `insert_edges_keyed_inner`'s extraction and
/// reusing the SAME [`unwind_edge_query_keyed_param`] renderer as the serial
/// flush, so the emitted Cypher is byte-identical. Returns each window's
/// `(query, rows)` plus the total edge count.
#[allow(clippy::too_many_arguments)]
fn render_edge_queries(
src_label: &str,
edge_label: &str,
dst_label: &str,
src_keys: &[&str],
dst_keys: &[&str],
batches: &[RecordBatch],
batch_size: usize,
) -> Result<(Vec<(String, usize)>, usize)> {
if batches.is_empty() {
return Err(FalkorError::EmptyBatches);
}
if src_keys.is_empty() || dst_keys.is_empty() {
return Err(FalkorError::ColumnNotFound(
"at least one key per endpoint".into(),
));
}
let single_legacy = src_keys == ["_gar_id"] && dst_keys == ["_gar_id"];
let src_cols_wanted: Vec<&str> = if single_legacy {
vec!["src"]
} else {
src_keys.to_vec()
};
let dst_cols_wanted: Vec<&str> = if single_legacy {
vec!["dst"]
} else {
dst_keys.to_vec()
};
let mut out: Vec<(String, usize)> = Vec::new();
let mut total = 0usize;
let mut row_buf: Vec<EdgeRow> = Vec::with_capacity(batch_size);
for batch in batches {
let schema = batch.schema();
let src_key_arrs: Vec<&dyn Array> = src_cols_wanted
.iter()
.map(|name| {
batch
.column_by_name(name)
.map(|c| c.as_ref())
.ok_or_else(|| FalkorError::ColumnNotFound((*name).into()))
})
.collect::<Result<_>>()?;
let dst_key_arrs: Vec<&dyn Array> = dst_cols_wanted
.iter()
.map(|name| {
batch
.column_by_name(name)
.map(|c| c.as_ref())
.ok_or_else(|| FalkorError::ColumnNotFound((*name).into()))
})
.collect::<Result<_>>()?;
let prop_cols: Vec<(&str, &dyn Array)> = schema
.fields()
.iter()
.zip(batch.columns())
.filter(|(f, _)| {
let n = f.name().as_str();
!src_cols_wanted.contains(&n) && !dst_cols_wanted.contains(&n)
})
.map(|(f, col)| (f.name().as_str(), col.as_ref()))
.collect();
for row in 0..batch.num_rows() {
let src_id: Vec<String> = src_key_arrs
.iter()
.map(|arr| arrow_value_to_cypher(*arr, row))
.collect::<Result<_>>()?;
let dst_id: Vec<String> = dst_key_arrs
.iter()
.map(|arr| arrow_value_to_cypher(*arr, row))
.collect::<Result<_>>()?;
let props: Vec<(String, String)> = prop_cols
.iter()
.map(|(name, arr)| {
let val = arrow_value_to_cypher(*arr, row)?;
Ok((name.to_string(), val))
})
.collect::<Result<_>>()?;
row_buf.push(EdgeRow {
src_id,
dst_id,
props,
});
total += 1;
if row_buf.len() >= batch_size {
let q = unwind_edge_query_keyed_param(
src_label, edge_label, dst_label, src_keys, dst_keys, &row_buf,
);
out.push((q, row_buf.len()));
row_buf.clear();
}
}
}
if !row_buf.is_empty() {
let q = unwind_edge_query_keyed_param(
src_label, edge_label, dst_label, src_keys, dst_keys, &row_buf,
);
out.push((q, row_buf.len()));
row_buf.clear();
}
Ok((out, total))
}
#[cfg(test)]
mod render_tests {
//! Connection-free proofs for the parallel bulk path. The rendered queries
//! FULLY determine what the parallel dispatch inserts (`dispatch_par` only
//! runs them concurrently), and the serial flush shares `build_vertex_query`
//! / `unwind_edge_query_keyed_param` — and now `push_vertex_row` / `resolve_cols` —
//! with these renderers, so byte-identical Cypher here is byte-identical
//! inserts on the wire, without a live DB.
//!
//! That sharing used to be a CLAIM: the doc comments said the two paths emit
//! byte-identical Cypher, but only `build_vertex_query` was actually shared —
//! the row loop was written out twice and only the parallel copy was tested.
//! `single_vertex_row_writer` is what makes the claim enforceable.
//!
//! RED-when-broken: exact-string + exact-count asserts on the renderers.
use super::*;
use arrow_array::{Int64Array, StringArray};
use arrow_schema::{DataType, Field, Schema};
use std::sync::Arc;
fn people(names: &[&str], ages: &[i64]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("name", DataType::Utf8, false),
Field::new("age", DataType::Int64, false),
])),
vec![
Arc::new(StringArray::from(names.to_vec())),
Arc::new(Int64Array::from(ages.to_vec())),
],
)
.unwrap()
}
/// The vertex row shape, asserted EXACTLY at its single writer — separator,
/// `_gar_id` carrier, key quoting and column order.
///
/// The serial path's row loop had no test at all: dropping `_gar_id` from its
/// call site left the whole 42-test suite green while `insert_vertices` stopped
/// emitting the property `insert_edges` matches endpoints by.
#[test]
fn vertex_row_shape_is_exact_at_its_single_writer() {
let batch = people(&["alice", "bob"], &[30, 41]);
let cols = resolve_cols(&batch);
let mut maps = String::new();
// First row of a window: no leading separator, `_gar_id` first.
push_vertex_row(&mut maps, &cols, 0, 0, 0).unwrap();
assert_eq!(maps, "{`_gar_id`: 0, `name`: 'alice', `age`: 30}");
// Second row: separator, and the id is the ABSOLUTE index, not the
// in-window one — this is what makes the parallel dispatch safe.
push_vertex_row(&mut maps, &cols, 1, 7, 1).unwrap();
assert_eq!(
maps,
"{`_gar_id`: 0, `name`: 'alice', `age`: 30}, \
{`_gar_id`: 7, `name`: 'bob', `age`: 41}"
);
}
/// **One writer, or the two dispatch paths drift again.**
///
/// The serial and parallel vertex loaders each used to build the row map
/// themselves. Nothing compared them, so the shapes could diverge silently —
/// and did not, only by luck. Now `push_vertex_row` is the single writer, and
/// this keeps it that way: a second `write_row_map` call site in this file is
/// a second definition of what a vertex looks like on the wire.
#[test]
fn single_vertex_row_writer() {
let src = include_str!("loader.rs");
// Needles are assembled at RUNTIME so this test's own source text cannot
// match them — a self-counting guard drifts the moment it is edited.
let writer = format!("{}{}", "write_row_map", "(");
let call = format!(
"{}{}",
"push_vertex_row(&mut maps, &cols, row, ", "total, pending)?;"
);
let writes = src.match_indices(&writer).count();
assert_eq!(
writes, 1,
"`write_row_map` must be called from exactly ONE place (`push_vertex_row`); \
found {writes} call sites in loader.rs. A second call site is a second \
definition of what a vertex looks like on the wire — the exact drift that \
let the serial path lose `_gar_id` with all 42 tests still green."
);
let calls = src.match_indices(&call).count();
assert_eq!(
calls, 2,
"both the serial (`insert_vertices_multi_inner`) and the parallel \
(`render_vertex_queries`) loop must render rows through `push_vertex_row`; \
found {calls} of the 2 expected call sites."
);
}
fn edges(src: &[i64], dst: &[i64]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(src.to_vec())),
Arc::new(Int64Array::from(dst.to_vec())),
],
)
.unwrap()
}
/// Exact query string + `_gar_id` = absolute row index across batches. If the
/// parallel path diverged from the serial Cypher (different `_gar_id` scheme,
/// map shape, or CREATE clause) this string compare fails.
#[test]
fn vertex_render_exact_and_gar_ids_across_batches() {
let (queries, total) = render_vertex_queries(
&["Person"],
&[
people(&["alice", "bob"], &[30, 41]),
people(&["carol"], &[25]),
],
512,
)
.unwrap();
assert_eq!(total, 3, "three vertices across two batches");
assert_eq!(
queries.len(),
1,
"one flush window under the big batch_size"
);
assert_eq!(queries[0].1, 3, "the window carries all three rows");
let expected = "UNWIND [\
{`_gar_id`: 0, `name`: 'alice', `age`: 30}, \
{`_gar_id`: 1, `name`: 'bob', `age`: 41}, \
{`_gar_id`: 2, `name`: 'carol', `age`: 25}\
] AS row CREATE (n:`Person`) SET n = row";
assert_eq!(queries[0].0, expected);
}
/// The window split honors `batch_size` and `_gar_id`s stay absolute; the
/// render is deterministic (same input → identical queries). Parallelizing
/// safely REQUIRES both (order-independent CREATE over disjoint ids).
#[test]
fn vertex_windowing_and_determinism() {
let (queries, total) = render_vertex_queries(
&["Person"],
&[people(&["a", "b", "c", "d", "e"], &[1, 2, 3, 4, 5])],
2,
)
.unwrap();
assert_eq!(total, 5);
assert_eq!(
queries.iter().map(|(_, n)| *n).collect::<Vec<_>>(),
vec![2, 2, 1],
"windows of batch_size=2 then a remainder of 1"
);
assert!(
queries[2].0.contains("`_gar_id`: 4"),
"the last row keeps its absolute id"
);
let (again, _) = render_vertex_queries(
&["Person"],
&[people(&["a", "b", "c", "d", "e"], &[1, 2, 3, 4, 5])],
2,
)
.unwrap();
assert_eq!(
queries, again,
"render is deterministic — a prerequisite for safe concurrency"
);
}
/// Edge render: exact count, `batch_size` windowing, and the MATCH→CREATE
/// shape (the legacy `_gar_id` single-key path routes through here unchanged).
#[test]
fn edge_render_counts_and_shape() {
let (queries, total) = render_edge_queries(
"Person",
"KNOWS",
"Person",
&["_gar_id"],
&["_gar_id"],
&[edges(&[0, 1, 2], &[1, 2, 0])],
2,
)
.unwrap();
assert_eq!(total, 3, "three edges");
assert_eq!(
queries.iter().map(|(_, n)| *n).collect::<Vec<_>>(),
vec![2, 1],
"windows of 2 then 1"
);
assert!(
queries[0].0.contains("CREATE (a)-[r:`KNOWS`]->(b)"),
"the edge query MATCHes endpoints and CREATEs the relationship"
);
}
/// **Core-saturation proof, DB-free.** `buffered_dispatch` — the scheduler
/// `insert_*_par` route every window through — must run up to `concurrency`
/// tasks *simultaneously* (that IS the fix for the ~1-core one-RTT-at-a-time
/// stall). A fake `run_one` bumps a shared in-flight counter on entry, yields
/// (so the scheduler fills the buffer before any completes), records the peak,
/// then decrements. With 24 windows @ concurrency 8 the peak MUST reach 8.
/// RED-when-broken: if the dispatch ever regressed to a serial `for … await`,
/// only one task is ever in flight and `max_inflight` collapses to 1.
/// A current-thread runtime is used deliberately — it is exactly the
/// single-reactor / many-in-flight-commands model of `MultiplexedConnection`,
/// so the overlap proven here is the overlap that saturates the real link.
#[test]
fn buffered_dispatch_runs_windows_concurrently() {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
let concurrency = 8usize;
let windows = 24usize;
let inflight = Arc::new(AtomicUsize::new(0));
let max_inflight = Arc::new(AtomicUsize::new(0));
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
let mut progressed = 0usize;
rt.block_on(async {
let items: Vec<usize> = (0..windows).map(|i| i + 1).collect(); // rows = i+1
buffered_dispatch(
items,
concurrency,
|rows| {
let inflight = inflight.clone();
let max_inflight = max_inflight.clone();
async move {
let now = inflight.fetch_add(1, Ordering::SeqCst) + 1;
max_inflight.fetch_max(now, Ordering::SeqCst);
// Yield twice so buffer_unordered fills to `concurrency`
// (each buffered future parks at its first await) before
// any task completes — forcing real overlap.
tokio::task::yield_now().await;
tokio::task::yield_now().await;
inflight.fetch_sub(1, Ordering::SeqCst);
Ok::<usize, FalkorError>(rows)
}
},
|n| progressed += n,
)
.await
.unwrap();
});
assert_eq!(
max_inflight.load(Ordering::SeqCst),
concurrency,
"buffer_unordered must overlap exactly `concurrency` tasks — a serial \
dispatch would peak at 1 (the ~1-core stall this path fixes)"
);
assert_eq!(inflight.load(Ordering::SeqCst), 0, "every task drained");
assert_eq!(
progressed,
windows * (windows + 1) / 2,
"on_progress summed every window's rows exactly once"
);
}
/// A job's error must abort the dispatch and come back as the ORIGINAL
/// `FalkorError` value — the gatling engine carries errors as `anyhow`, so
/// this pins the downcast round-trip (a lost or re-wrapped error would
/// break every caller matching on the variant, e.g. `is_already_indexed`).
#[test]
fn dispatch_error_survives_the_engine() {
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
let mut progressed = 0usize;
let res = rt.block_on(buffered_dispatch(
vec![1usize, 2, 3],
2,
|n| async move {
if n == 2 {
Err(FalkorError::InvalidBatchSize)
} else {
Ok(n)
}
},
|n| progressed += n,
));
assert!(
matches!(res, Err(FalkorError::InvalidBatchSize)),
"the job's exact error variant must survive the engine, got {res:?}"
);
}
/// `concurrency = 0` is floored to 1 (never a divide-by-zero / empty-buffer
/// stall) and still runs every item exactly once.
#[test]
fn buffered_dispatch_floors_concurrency_and_runs_all() {
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
let mut progressed = 0usize;
rt.block_on(async {
buffered_dispatch(
vec![10usize, 20, 30],
0,
|rows| async move { Ok::<usize, FalkorError>(rows) },
|n| progressed += n,
)
.await
.unwrap();
});
assert_eq!(progressed, 60, "all three items ran even at concurrency 0");
}
/// Both renderers reject empty input loudly (never a silent zero-insert),
/// matching the serial methods' `EmptyBatches`.
#[test]
fn empty_batches_error_loudly() {
assert!(matches!(
render_vertex_queries(&["Person"], &[], 512),
Err(FalkorError::EmptyBatches)
));
assert!(matches!(
render_edge_queries("A", "R", "B", &["_gar_id"], &["_gar_id"], &[], 512),
Err(FalkorError::EmptyBatches)
));
}
/// `create_index`'s idempotency hinges on classifying exactly ONE server
/// error — "already indexed" — as success. This proves the classifier
/// matches that message (case/prefix-insensitively) and NOTHING else, so a
/// re-populate's redundant `CREATE INDEX` is swallowed while a genuine fault
/// still aborts the load. RED-when-broken: relax the matcher to swallow all
/// errors, or tighten it to miss FalkorDB's real wording, and this fails.
#[test]
fn already_indexed_matches_only_that_error() {
use redis::RedisError;
// The live FalkorDB regression message (from korp): must classify true.
let live = FalkorError::Redis(RedisError::from((
redis::ErrorKind::ExtensionError,
"index build failed",
"Attribute: '_gar_id' is already indexed".to_string(),
)));
assert!(
is_already_indexed(&live),
"the real 'already indexed' abort is swallowed"
);
// A wording/casing variant still counts.
let variant = FalkorError::Redis(RedisError::from((
redis::ErrorKind::ExtensionError,
"err",
"'_gar_id' is ALREADY INDEXED".to_string(),
)));
assert!(
is_already_indexed(&variant),
"case/prefix drift still classifies"
);
// An UNRELATED redis error must NOT be swallowed (would mask real faults).
let other = FalkorError::Redis(RedisError::from((
redis::ErrorKind::ExtensionError,
"syntax error near ON",
)));
assert!(
!is_already_indexed(&other),
"unrelated redis errors still propagate"
);
// A non-redis FalkorError never matches.
assert!(!is_already_indexed(&FalkorError::EmptyBatches));
}
/// Auto-select policy: `GRAPH.BULK` is picked for (and only for) an EMPTY
/// graph — a fresh initial population. A non-empty graph must take the
/// incremental `UNWIND` path (bulk `CREATE`s, it can't MERGE onto existing
/// nodes). RED-when-broken: invert `choose_bulk` and the fast lane would
/// clobber a live graph with duplicates.
#[test]
fn bulk_chosen_only_for_empty_graph() {
assert!(choose_bulk(true), "empty graph → GRAPH.BULK fast lane");
assert!(!choose_bulk(false), "non-empty graph → UNWIND fallback");
}
/// `first_int` digs the `count(n)` scalar out of a nested GRAPH.QUERY reply
/// (result-set → row → cell) and out of a `BulkString` count, so
/// `graph_is_empty` reads `0` correctly across reply shapes.
#[test]
fn first_int_finds_count_scalar() {
use redis::Value;
let nested = Value::Array(vec![
Value::Array(vec![Value::BulkString(b"count(n)".to_vec())]),
Value::Array(vec![Value::Array(vec![Value::Int(0)])]),
]);
assert_eq!(first_int(&nested), Some(0));
let as_bulk = Value::Array(vec![Value::BulkString(b"7".to_vec())]);
assert_eq!(first_int(&as_bulk), Some(7));
assert_eq!(first_int(&Value::Array(vec![])), None);
}
/// The `db.indexes()` pre-check fires only when the reply mentions BOTH the
/// label and the property — so an existing `(Person, _gar_id)` index is
/// detected (skip the create) while a partial mention is not (fall through
/// to the create, whose catch is the backstop). RED-when-broken: match on
/// label-only or property-only and the negative cases flip.
#[test]
fn index_presence_needs_both_label_and_property() {
use redis::Value;
let bulk = |s: &str| Value::BulkString(s.as_bytes().to_vec());
// A db.indexes()-shaped reply: header, one row [label, [props…]], stats.
let reply = Value::Array(vec![
Value::Array(vec![bulk("label"), bulk("properties")]),
Value::Array(vec![Value::Array(vec![
bulk("Person"),
Value::Array(vec![bulk("_gar_id")]),
])]),
Value::Array(vec![bulk("Cached execution: 0")]),
]);
assert!(
reply_lists_index(&reply, "Person", "_gar_id"),
"both present → indexed"
);
assert!(
!reply_lists_index(&reply, "Person", "name"),
"property absent → not indexed"
);
assert!(
!reply_lists_index(&reply, "Loan", "_gar_id"),
"label absent → not indexed"
);
assert!(
!reply_lists_index(&Value::Array(vec![]), "Person", "_gar_id"),
"empty reply (no indexes yet) → not indexed"
);
}
}
#[cfg(test)]
mod created_stats_tests {
//! The load-happened proofs. Before these, both edge paths returned the rows
//! they *sent*, so a `MATCH … CREATE` whose endpoints did not match reported
//! a complete load over a graph the server had written nothing into — the
//! live `korp_demo` shape (121 200 nodes, **0 edges**) with a GREEN status
//! row, and the reason a driver's `edges_loaded == edges_total` assertion
//! could not fail.
//!
//! Connection-free: `created_from_reply` is fed the exact reply shapes
//! FalkorDB returns, and `buffered_dispatch` is driven with a fake `run_one`,
//! so the whole chain is provable without a container.
//!
//! The reply shapes below are not guesses. Measured against a live FalkorDB
//! (`localhost/knut-falkordb-valkey9`) on 2026-08-01:
//!
//! ```text
//! > GRAPH.QUERY g "UNWIND [{s:1,d:2}] AS r MATCH (a:Person {_gar_id:r.s}),
//! (b:Person {_gar_id:r.d}) CREATE (a)-[:KNOWS]->(b)"
//! Relationships created: 1
//! Cached execution: 0
//! Query internal execution time: 0.295895 milliseconds
//!
//! > (the same query with four rows whose endpoints match NOTHING)
//! Cached execution: 0
//! Query internal execution time: 0.284544 milliseconds
//! ```
//!
//! So: the query SUCCEEDS, and the `Relationships created` line is simply
//! ABSENT rather than `: 0`. That is exactly the pair of facts that made rows
//! sent look like a load, and it confirms both halves of this module's design
//! — the line is parseable when present, and "absent" has to mean zero.
//!
//! # What is NOT the red-proof for this file
//!
//! An earlier version of this comment claimed: "restore `Ok(rows)` in place of
//! `created_from_reply` in `dispatch_par`/`flush_edges` and
//! `rows_sent_is_not_relationships_created` +
//! `dispatch_par_contract_counts_created_not_rows` both fail." **That claim
//! was false, and it was checked on 2026-08-01 by doing exactly it.** Both
//! production call sites were reverted to rows-sent semantics and the result
//! was:
//!
//! ```text
//! test loader::created_stats_tests::rows_sent_is_not_relationships_created ... ok
//! test loader::created_stats_tests::dispatch_par_contract_counts_created_not_rows ... ok
//! test result: ok. 49 passed; 0 failed
//! ```
//!
//! The reason is structural and worth remembering: neither test can see a
//! production call site. The first calls `created_from_reply` DIRECTLY, and
//! the second drives the generic `buffered_dispatch` with its own closure —
//! not `dispatch_par`. They prove the parser is right; they say nothing about
//! whether `flush_edges` and `dispatch_par` call it. A connection-free test
//! cannot reach either function, so no test in this module ever could.
//!
//! # The actual guard
//!
//! Provenance is enforced by the TYPE, in [`server_reported`]: a per-reply
//! count is a `Created`, whose field is private to that module, so the revert
//! above no longer compiles —
//!
//! ```text
//! error[E0308]: mismatched types
//! --> crates/graphar-falkordb/src/loader.rs:516:9
//! 516 | Ok(sent)
//! | ^^^----^ this argument influences the type of `Ok`
//! = note: expected `Result<Created, _>`, found `Result<usize, _>`
//! ```
//!
//! and neither does the obvious next edit:
//!
//! ```text
//! error[E0423]: cannot initialize a tuple struct which contains private fields
//! | ^^^^^ private field
//! ```
//!
//! `dispatch_par` additionally takes `Vec<String>` rather than
//! `Vec<(String, rows)>`, so the rows it used to return are not in its scope
//! at all. A compile error is the one guard no test can be blind to (LAW #5:
//! fix by construction, not by a watcher).
use super::*;
use redis::Value;
/// The stats block FalkorDB appends to every `GRAPH.QUERY` reply. `created`
/// is `None` for the zero case, because FalkorDB **omits** the line rather
/// than emitting `: 0` — the ambiguity `created_from_reply` has to resolve.
fn stats_reply(line: Option<&str>) -> Value {
let mut stats = Vec::new();
if let Some(l) = line {
stats.push(Value::BulkString(l.as_bytes().to_vec()));
}
stats.push(Value::BulkString(b"Cached execution: 0".to_vec()));
stats.push(Value::BulkString(
b"Query internal execution time: 0.482375 milliseconds".to_vec(),
));
// Shape of a no-RETURN query: [header, rows, stats].
Value::Array(vec![
Value::Array(vec![]),
Value::Array(vec![]),
Value::Array(stats),
])
}
#[test]
fn reads_the_created_count_off_the_reply() {
let r = stats_reply(Some("Relationships created: 4"));
assert_eq!(
created_from_reply(&r, StatKey::RelationshipsCreated)
.unwrap()
.get(),
4,
"the count comes from the server's statistics line"
);
let n = stats_reply(Some("Nodes created: 512"));
assert_eq!(
created_from_reply(&n, StatKey::NodesCreated).unwrap().get(),
512
);
// Not confusable: a nodes reply must not answer the relationships question.
assert_eq!(
created_from_reply(&n, StatKey::RelationshipsCreated)
.unwrap()
.get(),
0,
"a reply that created only nodes created zero relationships"
);
}
/// THE filed defect, replayed. Four rows dispatched, the server matched no
/// endpoint and so created nothing, and the query itself SUCCEEDED. The old
/// code returned 4 here.
#[test]
fn rows_sent_is_not_relationships_created() {
let rows_sent = 4usize;
let reply = stats_reply(None); // success, stats present, no rel line
let created = created_from_reply(&reply, StatKey::RelationshipsCreated)
.unwrap()
.get();
assert_eq!(
created, 0,
"a successful MATCH…CREATE that matched nothing created nothing"
);
assert_ne!(
created, rows_sent,
"the whole defect: 4 rows sent must not be reported as 4 edges loaded"
);
// …and the phase verdict refuses to call that a load.
let v = verdict(created, rows_sent, StatKey::RelationshipsCreated);
assert!(
matches!(
v,
Err(FalkorError::NothingCreated {
what: "relationships",
rows: 4
})
),
"0 created from 4 rows must be an error, not a green 4; got {v:?}"
);
}
/// The parallel path's contract, proven through the real `buffered_dispatch`
/// with a fake per-window runner. Three windows of 100 rows each; the server
/// reports creating 7, 0 and 3. Honest total: 10, not 300.
#[test]
fn dispatch_par_contract_counts_created_not_rows() {
let windows: Vec<(String, usize)> =
vec![("q1".into(), 100), ("q2".into(), 100), ("q3".into(), 100)];
let replies = std::collections::HashMap::from([
("q1", stats_reply(Some("Relationships created: 7"))),
("q2", stats_reply(None)),
("q3", stats_reply(Some("Relationships created: 3"))),
]);
let mut progress = 0usize;
let sum = futures::executor::block_on(buffered_dispatch(
windows,
2,
|(q, _rows)| {
let reply = replies[q.as_str()].clone();
async move {
created_from_reply(&reply, StatKey::RelationshipsCreated).map(Created::get)
}
},
&mut |n| progress += n,
))
.unwrap();
assert_eq!(sum, 10, "7+0+3 created — NOT the 300 rows dispatched");
assert_eq!(
progress, 10,
"the progress meter advances on created too, else a driver fills to \
100% against an empty graph"
);
}
/// A reply we cannot read is an ERROR, never a silent 0 and never rows-sent.
/// This is the distinction that makes "absent line ⇒ 0" safe.
#[test]
fn an_unreadable_reply_errors_rather_than_guessing() {
for (name, v) in [
("empty array", Value::Array(vec![])),
("bare OK", Value::SimpleString("OK".into())),
("integer", Value::Int(4)),
("nil", Value::Nil),
] {
let r = created_from_reply(&v, StatKey::RelationshipsCreated);
assert!(
matches!(r, Err(FalkorError::UnreadableStats("relationships"))),
"{name}: no statistics block must be UnreadableStats, got {r:?}"
);
}
// …but a stats block with no matching line is a legitimate zero.
assert_eq!(
created_from_reply(&stats_reply(None), StatKey::RelationshipsCreated)
.unwrap()
.get(),
0,
"stats present + line absent = a real 0 (FalkorDB omits zero lines)"
);
}
/// `verdict` is not merely always-red: a real load passes through untouched,
/// and an empty input (0 rows, 0 created) is not an error.
#[test]
fn verdict_passes_a_real_load_and_an_empty_one() {
assert_eq!(
verdict(1200, 1200, StatKey::RelationshipsCreated).unwrap(),
1200
);
assert_eq!(
verdict(3, 1200, StatKey::RelationshipsCreated).unwrap(),
3,
"partial loads are reported as they are, not rejected — the count is the signal"
);
assert_eq!(verdict(0, 0, StatKey::NodesCreated).unwrap(), 0);
}
#[test]
fn stats_line_shapes_recognised() {
for good in [
"Query internal execution time: 0.482375 milliseconds",
"Cached execution: 0",
"Relationships created: 4",
"Nodes created: 512",
"Labels added: 1",
"Properties set: 1024",
] {
assert!(
looks_like_stats_line(good),
"should be a stats line: {good}"
);
}
for bad in ["OK", "Person", "", "some:thing"] {
assert!(
!looks_like_stats_line(bad),
"should NOT be a stats line: {bad:?}"
);
}
}
}