crates/graphar-flight/src/falkor.rs
use arrow_array::RecordBatch;
use arrow_schema::SchemaRef;
use redis::aio::MultiplexedConnection;
use crate::{
compact::{CompactIdNames, compact_response_to_verbose},
convert::{response_to_batch, response_to_batch_auto},
error::{FlightSqlError, Result},
};
/// The mutual-TLS client identity. It moved down into `graphar-falkordb` so the loader (the
/// write side) and this executor (the read side) share one type and one client builder; this
/// re-export keeps `graphar_flight::FalkorTls` working for every caller.
pub use graphar_falkordb::FalkorTls;
/// Executes Cypher queries against FalkorDB and converts results to Arrow.
///
/// `MultiplexedConnection` is `Clone + Send + Sync`, so this struct is cheap
/// to clone per-request — no mutex needed.
#[derive(Clone)]
pub struct FalkorExecutor {
conn: MultiplexedConnection,
graph: String,
/// Ask the server for `--compact` replies (ordinal ids + type-tagged cells
/// instead of repeated name strings) and rebuild the verbose tree client
/// side. **Off by default, and measured 2026-10-20 to be the right
/// default**: on this client stack (`redis` RESP2 `Value` tree + the
/// zero-loss verbose-tree rebuild) compact decoded 0.45× verbose on a
/// 100k-row scalar fetch and 0.77× on `RETURN n` — the type tags add more
/// RESP tokens than the dropped names save. See the benching doc.
compact: bool,
/// id→name catalogs for `--compact` decoding; refreshed on miss.
id_names: CompactIdNames,
}
impl FalkorExecutor {
/// Connect to FalkorDB over a **plain** `redis://` URL, no client identity.
/// This is the historical constructor; it now delegates to
/// [`connect_with_identity`](Self::connect_with_identity) with `None`.
pub async fn connect(url: &str, graph: impl Into<String>) -> Result<Self> {
Self::connect_with_identity(url, graph, None).await
}
/// Connect to FalkorDB, optionally presenting a **mutual-TLS client identity**.
///
/// * `tls == None` ⇒ the plain `redis::Client::open(url)` path (unchanged) — the
/// local-loopback demo container and every existing caller.
/// * `tls == Some(id)` ⇒ a `rediss://` mutual-TLS client built via
/// `redis::Client::build_with_tls`, presenting the vault-held client cert+key
/// and verifying the server against the supplied CA (or the system roots). The
/// thin client uses this to reach a remote FalkorDB with its stored identity.
///
/// The identity bytes are consumed here and never retained.
pub async fn connect_with_identity(
url: &str,
graph: impl Into<String>,
tls: Option<FalkorTls>,
) -> Result<Self> {
let client = graphar_falkordb::tls::client(url, tls)?;
let conn = client.get_multiplexed_async_connection().await?;
Ok(Self {
conn,
graph: graph.into(),
compact: false,
id_names: CompactIdNames::default(),
})
}
/// Opt into `GRAPH.QUERY … --compact` replies. Every query door below then
/// receives the compact wire and rebuilds the exact verbose `Value` tree
/// before the (unchanged) decoders run, so callers see byte-identical
/// `RecordBatch`es (pinned live by `integration_compact`).
///
/// **Leave it off for throughput.** Measured 2026-10-20 (release, 100k
/// rows, live server): scalar fetch 146.8 ms verbose vs 327.0 ms compact;
/// `RETURN n` 1087 ms vs 1410 ms. Scalar rows carry no repeated names, so
/// compact's `[type, value]` framing is strictly more wire; on entity rows
/// the saved name bytes did not pay for the extra RESP tokens plus the
/// rebuild. The flag exists because it is the server's lighter serializer
/// and a future direct compact→Arrow decoder could revisit the verdict.
pub fn with_compact(mut self, on: bool) -> Self {
self.compact = on;
self
}
/// Whether this executor asks for `--compact` replies.
pub fn compact(&self) -> bool {
self.compact
}
/// **The one place a Cypher query hits the wire.** Sends `command` (always
/// `GRAPH.QUERY`/`GRAPH.RO_QUERY`), appending `--compact` when enabled, and
/// returns the reply in VERBOSE shape either way. A compact reply that
/// names an uncached ordinal triggers one catalog refresh
/// (`CALL db.propertyKeys()` / `db.labels()` / `db.relationshipTypes()`)
/// and a single retry of the conversion — the reply itself is not re-fetched.
async fn graph_reply(&mut self, command: &str, cypher: &str) -> Result<redis::Value> {
let mut cmd = redis::cmd(command);
cmd.arg(&self.graph).arg(cypher);
if self.compact {
cmd.arg("--compact");
}
let raw: redis::Value = cmd.query_async(&mut self.conn).await?;
if !self.compact {
return Ok(raw);
}
match compact_response_to_verbose(&raw, &self.id_names) {
Err(FlightSqlError::CompactIdUnknown { .. }) => {
self.refresh_id_names().await?;
compact_response_to_verbose(&raw, &self.id_names)
}
done => done,
}
}
/// Re-read the three ordinal catalogs compact replies index into. Plain
/// (non-compact) queries on purpose: the catalogs are the decode input.
async fn refresh_id_names(&mut self) -> Result<()> {
self.id_names = CompactIdNames {
property_keys: self.catalog("CALL db.propertyKeys()").await?,
labels: self.catalog("CALL db.labels()").await?,
relationship_types: self.catalog("CALL db.relationshipTypes()").await?,
};
Ok(())
}
/// One catalog: rows of single string cells, ordinal position = id.
async fn catalog(&mut self, call: &str) -> Result<Vec<String>> {
let raw: redis::Value = redis::cmd("GRAPH.QUERY")
.arg(&self.graph)
.arg(call)
.query_async(&mut self.conn)
.await?;
let mut out = Vec::new();
if let redis::Value::Array(outer) = &raw
&& outer.len() >= 2
&& let redis::Value::Array(rows) = &outer[1]
{
out.reserve(rows.len());
for row in rows {
let cell = match row {
redis::Value::Array(cells) => cells.first(),
_ => None,
};
match cell {
Some(redis::Value::BulkString(b)) => {
out.push(String::from_utf8_lossy(b).into_owned());
}
Some(redis::Value::SimpleString(s)) => out.push(s.clone()),
other => {
return Err(FlightSqlError::MalformedResponse(format!(
"{call}: expected a string catalog cell, got {other:?}"
)));
}
}
}
}
Ok(out)
}
/// Run a read Cypher query; convert the response using the given schema.
pub async fn query(&mut self, cypher: &str, schema: &SchemaRef) -> Result<RecordBatch> {
let raw = self.graph_reply("GRAPH.QUERY", cypher).await?;
response_to_batch(raw, schema)
}
/// Run a read Cypher query using the auto-string schema fallback.
///
/// Column names are taken from the FalkorDB response header; all values
/// are returned as `Utf8`. No schema pre-registration is required.
/// [`Self::query`] as `GRAPH.RO_QUERY`: the server refuses any write in `cypher`, so a
/// read endpoint cannot be turned into a writer by what it is asked - and the ACL user
/// behind it needs `+graph.ro_query` alone.
pub async fn ro_query(&mut self, cypher: &str, schema: &SchemaRef) -> Result<RecordBatch> {
let raw = self.graph_reply("GRAPH.RO_QUERY", cypher).await?;
response_to_batch(raw, schema)
}
pub async fn query_auto(&mut self, cypher: &str) -> Result<RecordBatch> {
let raw = self.graph_reply("GRAPH.QUERY", cypher).await?;
response_to_batch_auto(raw)
}
/// Run a write Cypher query (CREATE / MERGE / DELETE …) with no result rows.
/// Returns the number of nodes/relationships affected (from the stats line).
pub async fn execute(&mut self, cypher: &str) -> Result<u64> {
let raw = self.graph_reply("GRAPH.QUERY", cypher).await?;
Ok(parse_affected_count(raw))
}
pub fn graph(&self) -> &str {
&self.graph
}
/// **Read one FalkorDB server config value as an integer**
/// (`GRAPH.CONFIG GET <key>`), e.g. `NODE_CREATION_BUFFER`.
///
/// This is a SERVER-level command like `GRAPH.LIST` — it touches no graph key, so
/// the executor's graph name is irrelevant. It lives here for the same reason
/// `graph_names` does: a caller that wanted it used to open a redis client of its
/// own, and then korp spoke the FalkorDB wire protocol in three places instead of
/// none.
///
/// FalkorDB answers either a flat `[name, value]` pair or a nested array of them;
/// both shapes are read.
pub async fn config_get_int(&mut self, key: &str) -> Result<i64> {
let raw: redis::Value = redis::cmd("GRAPH.CONFIG")
.arg("GET")
.arg(key)
.query_async(&mut self.conn)
.await?;
config_int(&raw).ok_or_else(|| {
crate::error::FlightSqlError::MalformedResponse(format!(
"could not parse {key} from GRAPH.CONFIG reply: {raw:?}"
))
})
}
/// **Delete this executor's graph key** (`GRAPH.DELETE`).
///
/// The counterpart to the fact that FalkorDB CREATES a graph key on the first
/// `GRAPH.QUERY` — a caller that made one (a test, a scratch import) needs a way
/// to take it away again, and it belongs on the one client rather than in a
/// second redis connection somewhere else. Deleting a key that does not exist is
/// an error from the engine, so callers that do not care use `let _ =`.
pub async fn delete_graph(&mut self) -> Result<()> {
redis::cmd("GRAPH.DELETE")
.arg(&self.graph)
.query_async::<redis::Value>(&mut self.conn)
.await?;
Ok(())
}
/// **Every graph on this endpoint** (`GRAPH.LIST`), sorted and de-duplicated.
///
/// `GRAPH.LIST` is a SERVER-level command, not a per-graph `GRAPH.QUERY`, so it
/// answers for the whole endpoint regardless of which graph this executor was
/// opened on — connect with any name (`""` is fine) to enumerate.
///
/// It runs on the connection this executor ALREADY holds. That is the point: a
/// caller that wanted the graph list used to open a second `redis::Client` of its
/// own, which made two independent FalkorDB clients in the constellation. There is
/// one, and it is this file.
pub async fn graph_names(&mut self) -> Result<Vec<String>> {
let mut graphs: Vec<String> = redis::cmd("GRAPH.LIST")
.query_async(&mut self.conn)
.await?;
graphs.sort();
graphs.dedup();
Ok(graphs)
}
}
/// Pull the integer out of a `GRAPH.CONFIG GET <k>` reply. FalkorDB answers either a
/// flat `[name, value]` pair or a nested `[[name, value], …]`; both are read, and a
/// value that arrives as a bulk string (`b"512"`) is parsed too.
fn config_int(v: &redis::Value) -> Option<i64> {
match v {
redis::Value::Int(n) => Some(*n),
redis::Value::BulkString(b) => String::from_utf8_lossy(b).trim().parse().ok(),
redis::Value::Array(items) => {
// `[name, value]` — take the value; otherwise recurse into each element.
if items.len() == 2
&& let Some(n) = config_int(&items[1])
{
return Some(n);
}
items.iter().find_map(config_int)
}
_ => None,
}
}
/// Pull a rough "nodes created + relationships created" count from the stats
/// array that FalkorDB appends to every response.
fn parse_affected_count(response: redis::Value) -> u64 {
let outer = match response {
redis::Value::Array(v) => v,
_ => return 0,
};
// Stats are in the last element.
let stats = match outer.last() {
Some(redis::Value::Array(s)) => s,
_ => return 0,
};
let mut total = 0u64;
for stat in stats {
if let redis::Value::BulkString(b) = stat {
let s = String::from_utf8_lossy(b);
// e.g. "Nodes created: 3" or "Relationships created: 1"
if (s.contains("created:") || s.contains("deleted:"))
&& let Some(n) = s
.split(':')
.nth(1)
.and_then(|p| p.trim().parse::<u64>().ok())
{
total += n;
}
}
}
total
}