crates/knut-graphdb/src/falkor.rs
//! The **FalkorDB** backend — emits Cypher and runs it through graphar-flight's
//! `FalkorExecutor` (Cypher in, Arrow out). This is the default backend; its
//! behaviour is byte-identical to querying FalkorDB directly, so a consumer's
//! stock build is unchanged by the abstraction.
//!
//! It is a CLIENT over ONE endpoint: [`FalkorBackend::new`] takes one string and
//! nothing else. It does not start FalkorDB, does not probe for one, and imports no
//! subprocess, no runtime and no orchestrator.
//!
//! **This crate holds the only path to a FalkorDB connection in the constellation.**
//! Every dial here goes through `graphar_flight::FalkorExecutor`, which is the single
//! file that constructs a `redis::Client` — including `GRAPH.LIST`, which used to be
//! a second, independent redis client on korp's side.
//!
//! The canonical query shapes live HERE (one source of truth, no per-consumer twin):
//! the `RETURN` tail, the whole-graph scene, the Person-anchored default scene, the
//! neighbour-expansion query and the shortest-path query.
use std::time::Duration;
use arrow_array::RecordBatch;
use arrow_array::{Array, StringArray};
use async_trait::async_trait;
/// The ONE integer-column reader, consumed from graphar-flight (L5) — it lives
/// beside the type sniff whose `Utf8` zero-row fallback it compensates for, so
/// the coercion table can never drift from the sniff that produces the columns.
/// See `graphar_flight::convert::i64_column` for the full why.
use graphar_flight::i64_column as i64_col;
// …and the ONE float-column reader, for exactly the same reason: FalkorDB hands a node's
// `lon`/`lat` over RESP2 as BULK STRINGS, which the sniff types `Utf8`, so a plain
// `Float64Array` downcast reads every live coordinate as absent.
use graphar_flight::f64_column as f64_col;
use crate::{
Cypher, CypherSupport, GraphDb, GraphEdge, GraphError, GraphNode, GraphSchema, SubGraph,
UpsertEdge, UpsertNode, cap,
};
/// The canonical `RETURN` tail every korp graph query shares — the aliases the
/// batch parser consumes: integer `src`/`dst` endpoints, optional `rel`/`sl`/`dl`
/// label strings, the optional **`sk`/`dk` natural-key** columns (a Vessel's
/// `mmsi`, an Aircraft's `icao24`, a Company's `org_nr`) the consumer resolves to
/// an object identity, and the optional **`slon`/`slat`/`dlon`/`dlat` PLACE**
/// columns.
///
/// # Why the place columns are here and not at the call site
///
/// A node that has a real position on the earth (a road junction, a weather station)
/// stores it as plain `lon`/`lat` properties, and the consumer wants to *draw the
/// graph where it is* rather than as a force blob. Those four aliases and
/// [`parse_batch`], the only thing that reads them, live in the same file for the
/// reason the rest of this tail does: a `RETURN` shape and its reader that live apart
/// drift apart. A node without the properties answers `null` and the place is simply
/// absent — every existing graph reads exactly as before.
pub const RETURN_TAIL: &str = "RETURN id(a) AS src, id(b) AS dst, type(r) AS rel, \
labels(a)[0] AS sl, labels(b)[0] AS dl, \
coalesce(a.mmsi, a.icao24, a.org_nr) AS sk, coalesce(b.mmsi, b.icao24, b.org_nr) AS dk, \
a.lon AS slon, a.lat AS slat, b.lon AS dlon, b.lat AS dlat";
/// The default **Person-anchored, bounded, ordered** scene — a legible
/// Person-centric graph, not a hairball sampled from a multi-million-row store.
/// `ORDER BY id` makes it deterministic (the same scene every run).
#[must_use]
pub fn scene_cypher(limit: usize) -> String {
format!(
"MATCH (a:Person)-[r]->(b) WITH a, r, b ORDER BY id(a), id(b) LIMIT {limit} {RETURN_TAIL}"
)
}
/// The **whole-graph** scene — every relationship, capped. This is what
/// [`GraphDb::read`] runs: a graph name and a cap, no caller-authored Cypher, which
/// is what keeps the base trait query-language-free.
///
/// The cap is FORMATTED IN rather than appended by [`ensure_limit`]. That matters:
/// a constant query text that already carried `LIMIT 300` made `ensure_limit` a
/// no-op, so `read(g, 7)` silently returned 300 rows. The cap the caller asked for
/// is the cap that reaches the engine.
///
/// It carries the four **place** columns of [`RETURN_TAIL`] (and not its natural-key
/// pair): this is the query that opens a graph in an explorer, so it is exactly the
/// read that has to know whether the graph can be drawn where it is. A graph whose
/// nodes have no `lon`/`lat` answers `null` four times and nothing else changes.
#[must_use]
pub fn whole_graph_cypher(limit: usize) -> String {
format!(
"MATCH (a)-[r]->(b) RETURN id(a) AS src, id(b) AS dst, type(r) AS rel, \
labels(a)[0] AS sl, labels(b)[0] AS dl, \
a.lon AS slon, a.lat AS slat, b.lon AS dlon, b.lat AS dlat LIMIT {limit}"
)
}
/// **Neighbour expansion (L-3)** — one node's 1-hop links (undirected, so both
/// in- and out-edges of `node_id` come back), capped at `limit` rows.
#[must_use]
pub fn neighbours_cypher(node_id: i64, limit: usize) -> String {
format!("MATCH (a)-[r]-(b) WHERE id(a) = {node_id} {RETURN_TAIL} LIMIT {limit}")
}
/// **Path finding (L-4)** — the shortest path between two nodes (≤ `max_hops`),
/// unwound into one row per edge in the canonical shape. `max_hops` floors at 1.
#[must_use]
pub fn path_cypher(from_id: i64, to_id: i64, max_hops: usize) -> String {
let hops = max_hops.max(1);
format!(
"MATCH (s), (t) WHERE id(s) = {from_id} AND id(t) = {to_id} \
MATCH p = shortestPath((s)-[*..{hops}]-(t)) \
UNWIND relationships(p) AS r \
WITH startNode(r) AS a, endNode(r) AS b, r {RETURN_TAIL}"
)
}
/// **Result-limit guard for big graphs.** A whole-graph `MATCH … RETURN …` with no
/// `LIMIT` can stream back millions of rows and wedge a viewer. `ensure_limit`
/// appends `LIMIT <cap>` to a query that carries no `LIMIT` clause of its own, so an
/// analyst's ad-hoc query is bounded by default; a query that already caps itself is
/// returned unchanged (its own limit wins). A `cap` of 0 is "no guard". A trailing
/// `;` is trimmed so the appended clause is well-formed.
#[must_use]
pub fn ensure_limit(cypher: &str, cap: usize) -> String {
let trimmed = cypher.trim().trim_end_matches(';').trim_end();
if cap == 0 || trimmed.is_empty() || has_limit_keyword(trimmed) {
return trimmed.to_string();
}
format!("{trimmed} LIMIT {cap}")
}
/// Whole-word, case-insensitive `LIMIT` detection — so a property named `limits` or
/// a label `Limit` does not fool the guard into thinking the query is bounded.
fn has_limit_keyword(cypher: &str) -> bool {
cypher
.to_ascii_lowercase()
.split(|c: char| !c.is_ascii_alphanumeric() && c != '_')
.any(|tok| tok == "limit")
}
/// **`host:port` ⇒ `redis://host:port`.** A consumer's per-backend config carries an
/// `endpoint = host:port`, while `graphar-flight` and the `redis` client both want a
/// URL. A string that already carries a scheme is passed through untouched — so a
/// `rediss://` (mutual-TLS) endpoint is never rewritten to plaintext.
#[must_use]
pub fn normalise_endpoint(endpoint: &str) -> String {
let e = endpoint.trim();
if e.contains("://") {
e.to_string()
} else {
format!("redis://{e}")
}
}
/// The default upper bound on a single graph read, env-overridable via
/// `KORP_GRAPH_READ_TIMEOUT_MS`. Never zero: a parse to 0/invalid falls back to the
/// default so the bound always fires.
#[must_use]
pub fn default_read_timeout() -> Duration {
let ms = std::env::var("KORP_GRAPH_READ_TIMEOUT_MS")
.ok()
.and_then(|s| s.trim().parse::<u64>().ok())
.filter(|&n| n > 0)
.unwrap_or(30_000);
Duration::from_millis(ms)
}
/// The PEM bytes of a mutual-TLS client identity, held so the backend can rebuild a
/// fresh [`graphar_flight::FalkorTls`] per dial.
///
/// `FalkorTls` was once neither `Clone` nor `Debug` upstream ("handed in only
/// at the connect call site and dropped after"), and a BACKEND dials more than once —
/// so the bytes live in this deliberately quiet struct (no `Debug`, never logged) and
/// a freshly-built `FalkorTls` is handed to each connect and dropped with it.
#[derive(Clone)]
struct FalkorIdentity {
client_cert_pem: Vec<u8>,
client_key_pem: Vec<u8>,
ca_pem: Option<Vec<u8>>,
}
impl FalkorIdentity {
fn to_tls(&self) -> graphar_flight::FalkorTls {
graphar_flight::FalkorTls {
client_cert_pem: self.client_cert_pem.clone(),
client_key_pem: self.client_key_pem.clone(),
ca_pem: self.ca_pem.clone(),
}
}
}
/// A client over ONE FalkorDB endpoint. Cheap to construct and to clone — it holds
/// the URL, the time bound and (optionally) a client identity, and connects per-op.
///
/// ```no_run
/// # async fn demo() -> Result<(), knut_graphdb::GraphError> {
/// use knut_graphdb::{FalkorBackend, GraphDb};
/// let db = FalkorBackend::new("127.0.0.1:6379");
/// for name in db.graph_names().await? { // every graph on the endpoint
/// let schema = db.open(&name).await?; // its catalogue
/// let scene = db.read(&name, 300).await?; // its nodes + edges
/// println!("{name}: {} labels, {} nodes", schema.labels.len(), scene.nodes.len());
/// }
/// # Ok(()) }
/// ```
#[derive(Clone)]
pub struct FalkorBackend {
/// The normalised `redis://…` / `rediss://…` URL (see [`normalise_endpoint`]).
url: String,
/// Upper bound on any single op. Defaults to [`default_read_timeout`].
timeout: Duration,
/// An optional mutual-TLS client identity. `None` ⇒ the plain `redis://` path.
/// This is a CREDENTIAL, not a lifecycle knob — it changes how the backend DIALS
/// the endpoint, never whether the endpoint exists.
tls: Option<FalkorIdentity>,
}
impl std::fmt::Debug for FalkorBackend {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// Never print the identity bytes.
f.debug_struct("FalkorBackend")
.field("url", &self.url)
.field("timeout_ms", &self.timeout.as_millis())
.field("mtls", &self.tls.is_some())
.finish()
}
}
impl FalkorBackend {
/// **Take an endpoint. Nothing else.** Accepts either a bare `host:port` or a
/// full `redis://` / `rediss://` URL; see [`normalise_endpoint`].
#[must_use]
pub fn new(endpoint: impl AsRef<str>) -> Self {
Self {
url: normalise_endpoint(endpoint.as_ref()),
timeout: default_read_timeout(),
tls: None,
}
}
/// Present a mutual-TLS client identity on every dial — the `rediss://` path.
/// Consumed with the connection; the backend never logs it.
#[must_use]
pub fn with_identity(mut self, tls: graphar_flight::FalkorTls) -> Self {
self.tls = Some(FalkorIdentity {
client_cert_pem: tls.client_cert_pem,
client_key_pem: tls.client_key_pem,
ca_pem: tls.ca_pem,
});
self
}
/// Override the per-op time bound (default: [`default_read_timeout`]).
#[must_use]
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self
}
/// The per-op time bound in force.
#[must_use]
pub fn timeout(&self) -> Duration {
self.timeout
}
/// Whether this backend dials over mutual TLS.
#[must_use]
pub fn is_mtls(&self) -> bool {
self.tls.is_some()
}
fn tls(&self) -> Option<graphar_flight::FalkorTls> {
self.tls.as_ref().map(FalkorIdentity::to_tls)
}
/// Bound a future by this backend's timeout — a wedged endpoint yields
/// [`GraphError::Timeout`], never a hang.
async fn bounded<F, T>(&self, fut: F) -> Result<T, GraphError>
where
F: std::future::Future<Output = Result<T, GraphError>>,
{
match tokio::time::timeout(self.timeout, fut).await {
Ok(r) => r,
Err(_) => Err(GraphError::Timeout {
endpoint: self.url.clone(),
ms: self.timeout.as_millis(),
}),
}
}
/// Connect to one graph on this endpoint. The ONE dial site in this crate.
async fn connect(&self, graph: &str) -> Result<graphar_flight::FalkorExecutor, GraphError> {
graphar_flight::FalkorExecutor::connect_with_identity(&self.url, graph, self.tls())
.await
.map_err(|e| {
let how = if self.is_mtls() { " (mutual TLS)" } else { "" };
GraphError::Connect(format!("connect{how} {}: {e}", self.url))
})
}
/// **Delete a graph key on this endpoint** (`GRAPH.DELETE`).
///
/// Deliberately an INHERENT method, not a [`GraphDb`] one: the trait is the read
/// surface every backend must serve, and a store that cannot delete is still a
/// complete store. It exists because `open`'s list-before-probe rule is about a
/// key FalkorDB creates by itself — whoever creates one needs the one client to
/// be able to remove it, rather than opening a second connection to do so.
pub async fn delete_graph(&self, graph: &str) -> Result<(), GraphError> {
self.bounded(async {
let mut exec = self.connect(graph).await?;
exec.delete_graph()
.await
.map_err(|e| GraphError::Backend(format!("GRAPH.DELETE {graph}: {e}")))
})
.await
}
/// **Read one FalkorDB server config value as an integer**
/// (`GRAPH.CONFIG GET <key>`) — e.g. `NODE_CREATION_BUFFER`. Inherent, not on
/// [`GraphDb`]: it is a FalkorDB server knob, not something every graph backend
/// has. It exists here so a consumer never has to open its own redis connection
/// to ask.
pub async fn config_get_int(&self, key: &str) -> Result<i64, GraphError> {
self.bounded(async {
let mut exec = self.connect("").await?;
exec.config_get_int(key)
.await
.map_err(|e| GraphError::Backend(format!("GRAPH.CONFIG GET {key}: {e}")))
})
.await
}
/// Connect, run one Cypher, and parse the Arrow result into a [`SubGraph`].
async fn run(&self, graph: &str, cypher: &str) -> Result<SubGraph, GraphError> {
self.bounded(async {
let mut exec = self.connect(graph).await?;
let batch = exec
.query_auto(cypher)
.await
.map_err(|e| GraphError::Backend(e.to_string()))?;
Ok(parse_batch(&batch))
})
.await
}
}
#[async_trait]
impl GraphDb for FalkorBackend {
fn kind(&self) -> &'static str {
"falkordb"
}
fn endpoint(&self) -> &str {
&self.url
}
/// `GRAPH.LIST` over the endpoint, bounded.
///
/// **Reachable-and-empty is `Ok(vec![])`; unreachable is an `Err`.** Nothing in
/// this function turns a failure into a list.
///
/// The graph name handed to the executor is `""`: `GRAPH.LIST` is a SERVER-level
/// command and touches no graph key, so enumerating cannot create one.
async fn graph_names(&self) -> Result<Vec<String>, GraphError> {
self.bounded(async {
let mut exec = self.connect("").await?;
exec.graph_names()
.await
.map_err(|e| GraphError::Backend(format!("GRAPH.LIST: {e}")))
})
.await
}
/// The graph's live catalogue. A name that is NOT on this endpoint is refused
/// with [`GraphError::NotFound`] rather than answered with an empty schema.
///
/// **The existence check runs BEFORE the schema probe, and that order is load
/// bearing** — FalkorDB CREATES a graph key on the first `GRAPH.QUERY` against an
/// unknown name. See the crate docs.
async fn open(&self, graph: &str) -> Result<GraphSchema, GraphError> {
// LIST FIRST. Never probe an unknown name.
let names = self.graph_names().await?;
if !names.iter().any(|n| n == graph) {
return Err(GraphError::NotFound {
graph: graph.to_string(),
endpoint: self.url.clone(),
});
}
self.bounded(async {
let mut exec = self.connect(graph).await?;
async fn probe(
exec: &mut graphar_flight::FalkorExecutor,
cypher: &str,
) -> Result<RecordBatch, GraphError> {
exec.query_auto(cypher)
.await
.map_err(|e| GraphError::Backend(format!("{cypher}: {e}")))
}
let labels = string_column(
&probe(&mut exec, "CALL db.labels() YIELD label RETURN label").await?,
"label",
);
let relationship_types = string_column(
&probe(
&mut exec,
"CALL db.relationshipTypes() YIELD relationshipType RETURN relationshipType",
)
.await?,
"relationshipType",
);
let property_keys = string_column(
&probe(
&mut exec,
"CALL db.propertyKeys() YIELD propertyKey RETURN propertyKey",
)
.await?,
"propertyKey",
);
let node_count = scalar_count(&probe(&mut exec, "MATCH (n) RETURN count(n)").await?);
let edge_count =
scalar_count(&probe(&mut exec, "MATCH ()-[r]->() RETURN count(r)").await?);
Ok(GraphSchema {
graph: graph.to_string(),
labels,
relationship_types,
property_keys,
node_count,
edge_count,
}
.normalised())
})
.await
}
/// The whole-graph scene, capped at `limit` rows. Runs [`whole_graph_cypher`] —
/// the caller authors no Cypher, which is what keeps [`GraphDb`] query-language
/// free.
async fn read(&self, graph: &str, limit: usize) -> Result<SubGraph, GraphError> {
self.run(graph, &whole_graph_cypher(cap(limit))).await
}
async fn neighbours(
&self,
graph: &str,
node_id: i64,
limit: usize,
) -> Result<SubGraph, GraphError> {
self.run(graph, &neighbours_cypher(node_id, cap(limit))).await
}
async fn shortest_path(
&self,
graph: &str,
from_id: i64,
to_id: i64,
max_hops: usize,
) -> Result<SubGraph, GraphError> {
self.run(graph, &path_cypher(from_id, to_id, max_hops)).await
}
async fn upsert(
&self,
graph: &str,
nodes: &[UpsertNode],
edges: &[UpsertEdge],
) -> Result<u64, GraphError> {
self.bounded(async {
let mut exec = self.connect(graph).await?;
let mut written = 0u64;
// Nodes: MERGE on a stable `_gid` property (FalkorDB assigns its own
// internal id(); `_gid` keeps the caller's id addressable + idempotent).
for n in nodes {
let label = sanitize_label(&n.label);
let key_set = match &n.key {
Some(k) => format!(", n.nk = '{}'", escape(k)),
None => String::new(),
};
let cy = format!(
"MERGE (n:{label} {{_gid: {}}}) SET n._gid = {}{key_set}",
n.id, n.id
);
written += exec
.execute(&cy)
.await
.map_err(|e| GraphError::Backend(e.to_string()))?;
}
// Edges: MERGE both endpoints by `_gid`, then the relationship.
for e in edges {
let rel = sanitize_label(&e.rel);
let cy = format!(
"MERGE (a {{_gid: {}}}) MERGE (b {{_gid: {}}}) MERGE (a)-[:{rel}]->(b)",
e.src_id, e.dst_id
);
written += exec
.execute(&cy)
.await
.map_err(|e| GraphError::Backend(e.to_string()))?;
}
Ok(written)
})
.await
}
/// FalkorDB speaks openCypher — this is the backend that opts IN.
fn cypher_support(&self) -> CypherSupport {
CypherSupport::Cypher
}
}
#[async_trait]
impl Cypher for FalkorBackend {
/// Run the caller's own Cypher verbatim (the free-form query bar path), bounded
/// by [`ensure_limit`] so an unbounded ad-hoc query cannot wedge a viewer. An
/// empty query falls back to the Person-anchored [`scene_cypher`].
async fn cypher(
&self,
graph: &str,
query: &str,
limit: usize,
) -> Result<SubGraph, GraphError> {
let cypher = if query.trim().is_empty() {
scene_cypher(cap(limit))
} else {
ensure_limit(query, cap(limit))
};
self.run(graph, &cypher).await
}
}
/// Read a single named string column from an auto-schema (all-`Utf8`) result batch,
/// dropping nulls — the shape `CALL db.labels()` / `db.relationshipTypes()` /
/// `db.propertyKeys()` return. Falls back to the FIRST column when `name` is absent
/// (a procedure's single `YIELD` column).
fn string_column(batch: &RecordBatch, name: &str) -> Vec<String> {
let col = batch
.column_by_name(name)
.or_else(|| (batch.num_columns() > 0).then(|| batch.column(0)))
.and_then(|c| c.as_any().downcast_ref::<StringArray>());
let Some(a) = col else { return Vec::new() };
(0..a.len())
.filter(|&i| !a.is_null(i))
.map(|i| a.value(i).to_string())
.collect()
}
/// Parse a single scalar count from a one-cell result — `query_auto` returns it as a
/// `Utf8` string (e.g. `"4"`), so this parses the first cell to `u64` (0 on absent /
/// non-numeric, never a panic). Accepts an `Int64` cell too (belt-and-braces).
fn scalar_count(batch: &RecordBatch) -> u64 {
if batch.num_rows() == 0 || batch.num_columns() == 0 {
return 0;
}
let col = batch.column(0);
if let Some(a) = col.as_any().downcast_ref::<StringArray>()
&& !a.is_null(0)
{
return a.value(0).trim().parse::<u64>().unwrap_or(0);
}
i64_col(batch, batch.schema().field(0).name())
.and_then(|c| c.first().copied().flatten())
.map(|v| v.max(0) as u64)
.unwrap_or(0)
}
/// Escape a single-quoted Cypher string literal.
fn escape(s: &str) -> String {
s.replace('\\', "\\\\").replace('\'', "\\'")
}
/// Keep only `[A-Za-z0-9_]` for a label / relationship type (Cypher identifiers
/// are not quotable the way string literals are). An empty result ⇒ `N`.
fn sanitize_label(s: &str) -> String {
let cleaned: String = s
.chars()
.filter(|c| c.is_ascii_alphanumeric() || *c == '_')
.collect();
if cleaned.is_empty() {
"N".to_string()
} else {
cleaned
}
}
/// Build a [`SubGraph`] from a query result batch (the `src`/`dst`/`rel`/`sl`/`dl`
/// [+ `sk`/`dk`] convention). `query_auto` returns every column as `Utf8`, but we
/// accept the whole integer family too (and parse `Utf8` ids), so a `query()` that
/// used the strongly-typed path still parses. A zero-row result ⇒ an empty
/// (non-error) sub-graph — the empty-graph case, not a shape mistake.
pub fn parse_batch(batch: &RecordBatch) -> SubGraph {
let Some(src) = i64_col(batch, "src") else {
return SubGraph::default();
};
let Some(dst) = i64_col(batch, "dst") else {
return SubGraph::default();
};
let strcol = |name: &str| -> Option<&StringArray> {
batch
.column_by_name(name)
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
};
let rel = strcol("rel");
let sl = strcol("sl");
let dl = strcol("dl");
let sk = strcol("sk");
let dk = strcol("dk");
let (slon, slat) = (f64_col(batch, "slon"), f64_col(batch, "slat"));
let (dlon, dlat) = (f64_col(batch, "dlon"), f64_col(batch, "dlat"));
let cell = |col: Option<&StringArray>, r: usize| -> String {
col.filter(|a| !a.is_null(r))
.map(|a| a.value(r).to_string())
.unwrap_or_default()
};
let key_cell = |col: Option<&StringArray>, r: usize| -> Option<String> {
col.filter(|a| !a.is_null(r))
.map(|a| a.value(r).to_string())
};
// A PLACE is `(lon, lat)` only when BOTH columns answered AND both are finite and
// in range. A half-answer (`lon` present, `lat` null) or a NaN would otherwise
// become a coordinate at the prime meridian, and a graph drawn geographically
// would put that node in the Gulf of Guinea rather than admit it has no place.
let place_cell = |lon: &Option<Vec<Option<f64>>>, lat: &Option<Vec<Option<f64>>>, r: usize| -> Option<[f64; 2]> {
let (x, y) = (*lon.as_ref()?.get(r)?, *lat.as_ref()?.get(r)?);
let (x, y) = (x?, y?);
(x.is_finite() && y.is_finite() && (-180.0..=180.0).contains(&x) && (-90.0..=90.0).contains(&y))
.then_some([x, y])
};
let mut nodes: Vec<GraphNode> = Vec::new();
let mut edges: Vec<GraphEdge> = Vec::new();
let mut seen: std::collections::HashSet<i64> = std::collections::HashSet::new();
let push_node = |nodes: &mut Vec<GraphNode>,
seen: &mut std::collections::HashSet<i64>,
id: i64,
label: String,
key: Option<String>,
place: Option<[f64; 2]>| {
if seen.insert(id) {
nodes.push(GraphNode {
id,
label,
natural_key: key,
place,
});
}
};
for r in 0..batch.num_rows() {
let (s, d) = match (src.get(r).copied().flatten(), dst.get(r).copied().flatten()) {
(Some(s), Some(d)) => (s, d),
_ => continue,
};
push_node(&mut nodes, &mut seen, s, cell(sl, r), key_cell(sk, r), place_cell(&slon, &slat, r));
push_node(&mut nodes, &mut seen, d, cell(dl, r), key_cell(dk, r), place_cell(&dlon, &dlat, r));
edges.push(GraphEdge {
src_id: s,
dst_id: d,
rel: cell(rel, r),
});
}
SubGraph { nodes, edges }
}
#[cfg(test)]
mod tests {
use super::*;
use arrow_schema::DataType;
#[test]
fn builders_share_the_return_tail() {
assert!(neighbours_cypher(42, 100).contains(RETURN_TAIL));
assert!(neighbours_cypher(42, 100).contains("id(a) = 42"));
assert!(neighbours_cypher(42, 100).contains("LIMIT 100"));
assert!(path_cypher(7, 9, 4).contains(RETURN_TAIL));
assert!(path_cypher(7, 9, 4).contains("shortestPath((s)-[*..4]-(t))"));
assert!(
path_cypher(1, 2, 0).contains("[*..1]"),
"0 hops clamps to 1"
);
assert!(scene_cypher(1000).contains("MATCH (a:Person)"));
assert!(scene_cypher(1000).contains("LIMIT 1000"));
}
/// `string_column` reads a named all-`Utf8` column dropping nulls (the
/// `CALL db.labels()` shape), and `scalar_count` parses a one-cell `Utf8` count —
/// the two readers [`GraphDb::open`] builds a live schema from. RED-when-broken
/// against a null leak or a mis-parsed count. (Carried over from korp's
/// `graph.rs`, whose copies of both readers the fold deleted.)
#[test]
fn schema_readers_read_labels_and_counts() {
use arrow_schema::{Field, Schema};
use std::sync::Arc;
let labels_batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("label", DataType::Utf8, true)])),
vec![Arc::new(StringArray::from(vec![
Some("Person"),
None,
Some("Company"),
]))],
)
.unwrap();
assert_eq!(string_column(&labels_batch, "label"), vec!["Person", "Company"]);
// Absent name → falls back to the first (only) column.
assert_eq!(string_column(&labels_batch, "nope"), vec!["Person", "Company"]);
let count_batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("count(n)", DataType::Utf8, true)])),
vec![Arc::new(StringArray::from(vec!["4"]))],
)
.unwrap();
assert_eq!(scalar_count(&count_batch), 4);
// A zero-row count batch is 0, not a panic.
let empty = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("count(n)", DataType::Utf8, true)])),
vec![Arc::new(StringArray::from(Vec::<&str>::new()))],
)
.unwrap();
assert_eq!(scalar_count(&empty), 0);
}
/// `ensure_limit` bounds an unbounded query and respects one that bounds itself.
/// Carried over from korp with the function.
#[test]
fn ensure_limit_bounds_only_what_is_unbounded() {
let already = "MATCH (n) RETURN n limit 10";
assert_eq!(ensure_limit(already, 300), already);
assert_eq!(
ensure_limit("MATCH (n) RETURN n LIMIT 5", 300),
"MATCH (n) RETURN n LIMIT 5"
);
// A property named `limits` must NOT be mistaken for a LIMIT clause.
assert_eq!(
ensure_limit("MATCH (a) WHERE a.limits > 3 RETURN a", 50),
"MATCH (a) WHERE a.limits > 3 RETURN a LIMIT 50"
);
// cap == 0 disables the guard — "the caller already bounded this".
assert_eq!(ensure_limit("MATCH (n) RETURN n", 0), "MATCH (n) RETURN n");
// Trailing semicolon trimmed before the clause is appended.
assert_eq!(ensure_limit("MATCH (n) RETURN n;", 7), "MATCH (n) RETURN n LIMIT 7");
// Empty query stays empty.
assert_eq!(ensure_limit(" ", 10), "");
}
#[test]
fn parse_batch_builds_subgraph() {
use arrow_array::Int64Array;
use arrow_schema::{Field, Schema};
use std::sync::Arc;
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("rel", DataType::Utf8, false),
Field::new("sl", DataType::Utf8, false),
Field::new("dl", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![1, 1])),
Arc::new(Int64Array::from(vec![2, 3])),
Arc::new(StringArray::from(vec!["WORKS_AT", "LIVES_AT"])),
Arc::new(StringArray::from(vec!["Person", "Person"])),
Arc::new(StringArray::from(vec!["Company", "Address"])),
],
)
.unwrap();
let sub = parse_batch(&batch);
assert_eq!(sub.nodes.len(), 3);
assert_eq!(sub.edges.len(), 2);
assert_eq!(sub.nodes[0].label, "Person");
// A batch with no place columns leaves every node placeless — the graphs that
// are not anywhere must not acquire a coordinate.
assert!(sub.nodes.iter().all(|n| n.place.is_none()));
}
/// **The place columns reach the node** — and only when the row actually answered
/// both of them with a real coordinate.
///
/// Four rows, deliberately: one whole pair, one with a null `lat`, one with a NaN
/// `lon`, one with a longitude off the earth. Only the first is a place. The rest
/// are the ways a half-answer would otherwise become a node drawn in the Gulf of
/// Guinea, which is worse than a node with no place at all.
#[test]
fn a_place_needs_both_halves_finite_and_on_the_earth() {
use arrow_array::{Float64Array, Int64Array};
use arrow_schema::{Field, Schema};
use std::sync::Arc;
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
Field::new("slon", DataType::Float64, true),
Field::new("slat", DataType::Float64, true),
Field::new("dlon", DataType::Float64, true),
Field::new("dlat", DataType::Float64, true),
])),
vec![
Arc::new(Int64Array::from(vec![10, 20, 30, 40])),
Arc::new(Int64Array::from(vec![11, 21, 31, 41])),
Arc::new(Float64Array::from(vec![
Some(18.0686),
Some(13.0038),
Some(f64::NAN),
Some(999.0),
])),
Arc::new(Float64Array::from(vec![
Some(59.3293),
None,
Some(57.7089),
Some(67.8558),
])),
Arc::new(Float64Array::from(vec![None, None, None, None])),
Arc::new(Float64Array::from(vec![None, None, None, None])),
],
)
.unwrap();
let sub = parse_batch(&batch);
let by_id = |id: i64| sub.nodes.iter().find(|n| n.id == id).expect("node").place;
assert_eq!(by_id(10), Some([18.0686, 59.3293]), "a whole pair IS a place");
assert_eq!(by_id(20), None, "lon without lat is not a place");
assert_eq!(by_id(30), None, "NaN is not a place");
assert_eq!(by_id(40), None, "longitude 999 is not on the earth");
// The dst side answered null throughout: no dst node gains a place either.
assert!(sub.nodes.iter().filter(|n| n.id % 10 == 1).all(|n| n.place.is_none()));
}
/// **The shape a LIVE FalkorDB actually sends.** Not the shape a fixture is
/// convenient to build.
///
/// FalkorDB stores `lon`/`lat` as doubles, but over RESP2 they come back as bulk
/// strings, which the sniff types `Utf8`. The first cut of this parser downcast to
/// `Float64Array` and nothing else: every fixture passed, and on oden a 77 916-
/// junction Stockholm slice arrived with `geo_nodes: 0` and drew as a hairball —
/// twice — against a server that had answered perfectly. This is that batch.
#[test]
fn a_place_survives_the_utf8_the_sniff_gives_a_resp2_double() {
use arrow_array::Int64Array;
use arrow_schema::{Field, Schema};
use std::sync::Arc;
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
// Utf8, exactly as `sniff_column_types` leaves a RESP2 bulk string.
Field::new("slon", DataType::Utf8, true),
Field::new("slat", DataType::Utf8, true),
Field::new("dlon", DataType::Utf8, true),
Field::new("dlat", DataType::Utf8, true),
])),
vec![
Arc::new(Int64Array::from(vec![0i64, 1])),
Arc::new(Int64Array::from(vec![63i64, 73])),
Arc::new(StringArray::from(vec!["18.064568", "18.083780"])),
Arc::new(StringArray::from(vec!["59.359859", "59.349937"])),
Arc::new(StringArray::from(vec!["18.066515", "not a number"])),
Arc::new(StringArray::from(vec!["59.359802", "59.349957"])),
],
)
.unwrap();
let sub = parse_batch(&batch);
let by_id = |id: i64| sub.nodes.iter().find(|n| n.id == id).expect("node").place;
assert_eq!(by_id(0), Some([18.064568, 59.359859]), "a stringified double IS a place");
assert_eq!(by_id(63), Some([18.066515, 59.359802]));
assert_eq!(by_id(73), None, "an unparseable string is not a place");
assert_eq!(
sub.nodes.iter().filter(|n| n.place.is_some()).count(),
3,
"three of the four endpoints are placed"
);
}
/// The four place aliases are in the tail AND in the whole-graph scene — the two
/// reads that feed an explorer. A tail that carries them while the scene query
/// does not is exactly the drift this pins.
#[test]
fn every_scene_query_returns_the_place_columns() {
for q in [
RETURN_TAIL.to_string(),
whole_graph_cypher(300),
scene_cypher(50),
neighbours_cypher(7, 100),
path_cypher(1, 2, 4),
] {
for alias in ["AS slon", "AS slat", "AS dlon", "AS dlat"] {
assert!(q.contains(alias), "{alias} missing from: {q}");
}
}
}
#[test]
fn zero_row_utf8_is_empty_not_error() {
use arrow_schema::{Field, Schema};
use std::sync::Arc;
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("src", DataType::Utf8, true),
Field::new("dst", DataType::Utf8, true),
])),
vec![
Arc::new(StringArray::from(Vec::<&str>::new())) as _,
Arc::new(StringArray::from(Vec::<&str>::new())) as _,
],
)
.unwrap();
let sub = parse_batch(&batch);
assert!(sub.nodes.is_empty() && sub.edges.is_empty());
}
}