crates/graphar-flight/src/compact.rs
//! `GRAPH.QUERY … --compact` reply → the **verbose-equivalent** redis `Value`
//! tree, so the one existing decoder ([`crate::convert::response_to_batch_auto`]
//! / [`crate::convert::response_to_batch`]) keeps producing byte-identical
//! `RecordBatch`es whichever wire representation carried the result.
//!
//! ## Why
//!
//! Measured live (benching doc, 2026-10-19): on a 100k-row reply the RESP
//! **wire** costs 150–162 ms while the client decode costs 13 ms. `--compact`
//! replaces repeated property/label/relationship-type *names* with ordinal
//! **ids** and type-tagged `[type_id, value]` pairs — materially less RESP for
//! the server to serialize and the client to parse. This module is the decode
//! half: it rebuilds exactly what the verbose wire would have delivered, so
//! zero-loss is structural (the same `convert.rs` decoder runs either way).
//!
//! ## The compact wire shape (probed live 2026-10-20, FalkorDB/valkey-9)
//!
//! * header: `[[column_type:int, name], …]` (column_type is the legacy 1/2/3
//! scalar/node/relation marker; the per-cell type tag below supersedes it)
//! * row cell: `[value_type:int, payload]` with
//! `1 NULL, 2 STRING, 3 INTEGER, 4 BOOLEAN ("true"/"false" string over
//! RESP2), 5 DOUBLE (printed string over RESP2), 6 ARRAY (of cells),
//! 7 EDGE, 8 NODE, 9 PATH, 10 MAP (flat `[key, cell, …]`), 11 POINT
//! (`[lat, lon]` printed strings)`
//! * node payload: `[id:int, [label_id…], [[prop_id, value_type, value]…]]`
//! * edge payload: `[id:int, reltype_id:int, src_id:int, dst_id:int,
//! [[prop_id, value_type, value]…]]`
//!
//! ## The verbose shapes this reproduces (probed on the same server)
//!
//! * scalar string/int/bool/double cells arrive byte-identical in both modes
//! (RESP2 sends booleans and doubles as their printed strings either way)
//! * a top-level array/map/point cell arrives in verbose mode as its **printed
//! string** (`"[a, b]"`, `"{s: x, d: 2.500000}"`,
//! `"point({latitude:57.700001, longitude:11.900000})"`) — doubles inside
//! collections print C-`%f` style with 6 decimals, strings print unquoted,
//! null prints `NULL`
//! * a node arrives as `[["id", int], ["labels", [name…]],
//! ["properties", [[name, value]…]]]`
//! * an edge arrives as `[["id", int], ["type", name], ["src_node", int],
//! ["dest_node", int], ["properties", [[name, value]…]]]`
//!
//! Ids are ordinals into the server catalogs (`CALL db.propertyKeys()`,
//! `db.labels()`, `db.relationshipTypes()`); [`CompactIdNames`] caches them and
//! an unknown ordinal surfaces as [`FlightSqlError::CompactIdUnknown`], which
//! the query door treats as "refresh the catalogs and convert again".
//!
//! Honest limits: a `PATH` cell (value type 9) and a node/edge nested *inside*
//! a collection have no pinned verbose rendering here and are refused loudly
//! rather than guessed.
use std::fmt::Write as _;
use redis::Value;
use crate::error::{FlightSqlError, Result};
/// Cached id→name catalogs for `--compact` decoding, in server ordinal order.
#[derive(Debug, Clone, Default)]
pub struct CompactIdNames {
pub property_keys: Vec<String>,
pub labels: Vec<String>,
pub relationship_types: Vec<String>,
}
const T_NULL: i64 = 1;
const T_STRING: i64 = 2;
const T_INTEGER: i64 = 3;
const T_BOOLEAN: i64 = 4;
const T_DOUBLE: i64 = 5;
const T_ARRAY: i64 = 6;
const T_EDGE: i64 = 7;
const T_NODE: i64 = 8;
const T_PATH: i64 = 9;
const T_MAP: i64 = 10;
const T_POINT: i64 = 11;
fn malformed(what: &str, got: &Value) -> FlightSqlError {
FlightSqlError::MalformedResponse(format!("compact reply: {what}, got {got:?}"))
}
fn bulk(s: String) -> Value {
Value::BulkString(s.into_bytes())
}
fn pair(key: &str, v: Value) -> Value {
Value::Array(vec![bulk(key.to_string()), v])
}
/// Convert a whole `GRAPH.QUERY … --compact` reply into the `Value` tree the
/// **verbose** wire would have delivered, so the existing decoders apply
/// unchanged. A stats-only reply (a write with no `RETURN`) is identical in
/// both modes and passes through; stats arrays are passed through verbatim.
pub fn compact_response_to_verbose(resp: &Value, names: &CompactIdNames) -> Result<Value> {
let outer = match resp {
Value::Array(v) => v,
other => return Err(malformed("expected Array at top level", other)),
};
// Stats-only (write) replies are byte-identical in both modes.
if outer.len() < 2 {
return Ok(resp.clone());
}
let header = convert_header(&outer[0])?;
let rows = match &outer[1] {
Value::Array(rows) => Value::Array(
rows.iter()
.map(|row| {
let cells = match row {
Value::Array(c) => c,
other => return Err(malformed("expected Array row", other)),
};
Ok(Value::Array(
cells
.iter()
.map(|c| cell_to_verbose(c, names))
.collect::<Result<Vec<_>>>()?,
))
})
.collect::<Result<Vec<_>>>()?,
),
Value::Nil => Value::Nil,
other => return Err(malformed("expected Array for data rows", other)),
};
let mut out = Vec::with_capacity(outer.len());
out.push(header);
out.push(rows);
out.extend(outer.iter().skip(2).cloned()); // stats verbatim
Ok(Value::Array(out))
}
/// `[[column_type, name], …]` → `[name, …]` (the verbose header row).
fn convert_header(header: &Value) -> Result<Value> {
let cols = match header {
Value::Array(c) => c,
other => return Err(malformed("expected Array header", other)),
};
let names: Vec<Value> = cols
.iter()
.map(|col| match col {
Value::Array(kv) if kv.len() == 2 && matches!(kv[0], Value::Int(_)) => {
Ok(kv[1].clone())
}
other => Err(malformed(
"expected [column_type, name] header entry — is this really a --compact reply?",
other,
)),
})
.collect::<Result<_>>()?;
Ok(Value::Array(names))
}
/// Split a `[value_type, payload]` cell. A bare `[value_type]` (seen for NULL)
/// yields a `Nil` payload.
fn split_cell(cell: &Value) -> Result<(i64, &Value)> {
static NIL: Value = Value::Nil;
match cell {
Value::Array(kv) if !kv.is_empty() => match kv[0] {
Value::Int(t) => Ok((t, kv.get(1).unwrap_or(&NIL))),
_ => Err(malformed("expected Int value-type tag", cell)),
},
other => Err(malformed("expected [value_type, payload] cell", other)),
}
}
/// One result cell → its verbose-wire equivalent.
fn cell_to_verbose(cell: &Value, names: &CompactIdNames) -> Result<Value> {
let (t, payload) = split_cell(cell)?;
match t {
T_NULL => Ok(Value::Nil),
// RESP2 carries these identically in both modes: string bytes, raw
// integer, "true"/"false", the server-printed double string.
T_STRING | T_INTEGER | T_BOOLEAN | T_DOUBLE => Ok(payload.clone()),
// Verbose mode delivers collections and points as their printed string.
T_ARRAY | T_MAP | T_POINT => {
let mut s = String::new();
print_value(&mut s, cell, names)?;
Ok(bulk(s))
}
T_NODE => node_to_verbose(payload, names),
T_EDGE => edge_to_verbose(payload, names),
T_PATH => Err(malformed(
"PATH cells have no pinned verbose rendering here (unsupported)",
cell,
)),
_ => Err(malformed("unknown compact value type", cell)),
}
}
/// Print a compact cell exactly as FalkorDB's verbose formatter stringifies
/// values nested in collections: strings raw/unquoted, `NULL`, `%f`-style
/// doubles with 6 decimals, `[a, b]`, `{k: v}`, `point({latitude:…, longitude:…})`.
fn print_value(out: &mut String, cell: &Value, names: &CompactIdNames) -> Result<()> {
let (t, payload) = split_cell(cell)?;
match t {
T_NULL => out.push_str("NULL"),
T_STRING | T_BOOLEAN => out.push_str(&payload_str(payload)?),
T_INTEGER => match payload {
Value::Int(i) => {
let _ = write!(out, "{i}");
}
other => return Err(malformed("expected Int payload", other)),
},
T_DOUBLE => {
let x: f64 = payload_str(payload)?
.parse()
.map_err(|_| malformed("unparsable double payload", payload))?;
push_c_double(out, x);
}
T_ARRAY => {
let items = payload_array(payload)?;
out.push('[');
for (i, item) in items.iter().enumerate() {
if i > 0 {
out.push_str(", ");
}
print_value(out, item, names)?;
}
out.push(']');
}
T_MAP => {
// Flat `[key, cell, key, cell, …]`.
let items = payload_array(payload)?;
if items.len() % 2 != 0 {
return Err(malformed("expected alternating [key, cell] map payload", payload));
}
out.push('{');
for (i, kv) in items.chunks(2).enumerate() {
if i > 0 {
out.push_str(", ");
}
out.push_str(&payload_str(&kv[0])?);
out.push_str(": ");
print_value(out, &kv[1], names)?;
}
out.push('}');
}
T_POINT => {
let items = payload_array(payload)?;
if items.len() != 2 {
return Err(malformed("expected [lat, lon] point payload", payload));
}
let lat: f64 = payload_str(&items[0])?
.parse()
.map_err(|_| malformed("unparsable point latitude", payload))?;
let lon: f64 = payload_str(&items[1])?
.parse()
.map_err(|_| malformed("unparsable point longitude", payload))?;
out.push_str("point({latitude:");
push_c_double(out, lat);
out.push_str(", longitude:");
push_c_double(out, lon);
out.push_str("})");
}
T_NODE | T_EDGE | T_PATH => {
return Err(malformed(
"a graph entity nested inside a collection has no pinned verbose rendering (unsupported)",
cell,
));
}
_ => return Err(malformed("unknown compact value type", cell)),
}
Ok(())
}
/// C `printf("%f")`, which is what the verbose formatter uses for doubles
/// inside collections: 6 decimals, `inf`/`-inf`, lowercase `nan`.
fn push_c_double(out: &mut String, x: f64) {
if x.is_nan() {
out.push_str("nan");
} else if x.is_infinite() {
out.push_str(if x > 0.0 { "inf" } else { "-inf" });
} else {
let _ = write!(out, "{x:.6}");
}
}
fn payload_str(v: &Value) -> Result<String> {
match v {
Value::BulkString(b) => Ok(String::from_utf8_lossy(b).into_owned()),
Value::SimpleString(s) => Ok(s.clone()),
other => Err(malformed("expected string payload", other)),
}
}
fn payload_array(v: &Value) -> Result<&Vec<Value>> {
match v {
Value::Array(items) => Ok(items),
other => Err(malformed("expected Array payload", other)),
}
}
fn payload_int(v: &Value) -> Result<i64> {
match v {
Value::Int(i) => Ok(*i),
other => Err(malformed("expected Int payload", other)),
}
}
fn name_of<'a>(catalog: &'a [String], id: i64, kind: &'static str) -> Result<&'a str> {
usize::try_from(id)
.ok()
.and_then(|i| catalog.get(i))
.map(String::as_str)
.ok_or(FlightSqlError::CompactIdUnknown {
kind,
id,
have: catalog.len(),
})
}
/// `[[prop_id, value_type, value]…]` → verbose `[[name, value]…]`.
fn props_to_verbose(props: &Value, names: &CompactIdNames) -> Result<Value> {
let entries = payload_array(props)?;
let out: Vec<Value> = entries
.iter()
.map(|entry| {
let triple = payload_array(entry)?;
if triple.len() != 3 {
return Err(malformed("expected [prop_id, value_type, value] property", entry));
}
let pid = payload_int(&triple[0])?;
let name = name_of(&names.property_keys, pid, "property key")?;
let cell = Value::Array(vec![triple[1].clone(), triple[2].clone()]);
Ok(Value::Array(vec![
bulk(name.to_string()),
cell_to_verbose(&cell, names)?,
]))
})
.collect::<Result<_>>()?;
Ok(Value::Array(out))
}
/// Compact node payload → the verbose node tree.
fn node_to_verbose(payload: &Value, names: &CompactIdNames) -> Result<Value> {
let parts = payload_array(payload)?;
if parts.len() != 3 {
return Err(malformed("expected [id, labels, properties] node payload", payload));
}
let labels: Vec<Value> = payload_array(&parts[1])?
.iter()
.map(|lid| {
let id = payload_int(lid)?;
Ok(bulk(name_of(&names.labels, id, "label")?.to_string()))
})
.collect::<Result<_>>()?;
Ok(Value::Array(vec![
pair("id", parts[0].clone()),
pair("labels", Value::Array(labels)),
pair("properties", props_to_verbose(&parts[2], names)?),
]))
}
/// Compact edge payload → the verbose edge tree (note the server's verbose key
/// is `dest_node`, not `dst_node`).
fn edge_to_verbose(payload: &Value, names: &CompactIdNames) -> Result<Value> {
let parts = payload_array(payload)?;
if parts.len() != 5 {
return Err(malformed(
"expected [id, reltype, src, dst, properties] edge payload",
payload,
));
}
let tid = payload_int(&parts[1])?;
Ok(Value::Array(vec![
pair("id", parts[0].clone()),
pair(
"type",
bulk(name_of(&names.relationship_types, tid, "relationship type")?.to_string()),
),
pair("src_node", parts[2].clone()),
pair("dest_node", parts[3].clone()),
pair("properties", props_to_verbose(&parts[4], names)?),
]))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::convert::response_to_batch_auto;
fn cell(t: i64, v: Value) -> Value {
Value::Array(vec![Value::Int(t), v])
}
fn bs(s: &str) -> Value {
Value::BulkString(s.as_bytes().to_vec())
}
fn compact_header(names: &[&str]) -> Value {
Value::Array(
names
.iter()
.map(|n| Value::Array(vec![Value::Int(1), bs(n)]))
.collect(),
)
}
fn verbose_header(names: &[&str]) -> Value {
Value::Array(names.iter().map(|n| bs(n)).collect())
}
fn names() -> CompactIdNames {
CompactIdNames {
property_keys: vec!["i".into(), "s".into(), "f".into(), "b".into(), "k".into()],
labels: vec!["P".into(), "A".into(), "B".into()],
relationship_types: vec!["R".into()],
}
}
/// ZERO-LOSS pin, compact edition of `auto_decode_varied_fixture_pins_whole_batch`:
/// a compact reply with every scalar kind, a null, an array, a map and a
/// point converts to EXACTLY the verbose tree the live server sends over
/// RESP2 (shapes probed 2026-10-20) — and the one shared decoder then builds
/// the identical batch from both.
#[test]
fn compact_varied_fixture_matches_verbose_tree_and_batch() {
let compact = Value::Array(vec![
compact_header(&["id", "score", "ok", "name", "ghost", "ls", "m", "pt"]),
Value::Array(vec![Value::Array(vec![
cell(T_INTEGER, Value::Int(7)),
cell(T_DOUBLE, bs("0.5")),
cell(T_BOOLEAN, bs("true")),
cell(T_STRING, bs("alpha")),
cell(T_NULL, Value::Nil),
cell(
T_ARRAY,
Value::Array(vec![
cell(T_INTEGER, Value::Int(1)),
cell(T_STRING, bs("a")),
cell(T_DOUBLE, bs("1.5")),
cell(T_NULL, Value::Nil),
]),
),
cell(
T_MAP,
Value::Array(vec![
bs("s"),
cell(T_STRING, bs("x")),
bs("d"),
cell(T_DOUBLE, bs("2.5")),
bs("inner"),
cell(T_ARRAY, Value::Array(vec![cell(T_INTEGER, Value::Int(1))])),
]),
),
cell(
T_POINT,
Value::Array(vec![bs("57.7000007629395"), bs("11.8999996185303")]),
),
])]),
Value::Array(vec![bs("Query internal execution time: 0.1 milliseconds")]),
]);
// The verbose tree the SAME query delivers over RESP2, as probed live.
let verbose = Value::Array(vec![
verbose_header(&["id", "score", "ok", "name", "ghost", "ls", "m", "pt"]),
Value::Array(vec![Value::Array(vec![
Value::Int(7),
bs("0.5"),
bs("true"),
bs("alpha"),
Value::Nil,
bs("[1, a, 1.500000, NULL]"),
bs("{s: x, d: 2.500000, inner: [1]}"),
bs("point({latitude:57.700001, longitude:11.900000})"),
])]),
Value::Array(vec![bs("Query internal execution time: 0.1 milliseconds")]),
]);
let converted = compact_response_to_verbose(&compact, &names()).unwrap();
assert_eq!(converted, verbose, "compact must rebuild the exact verbose tree");
let from_compact = response_to_batch_auto(converted).unwrap();
let from_verbose = response_to_batch_auto(verbose).unwrap();
assert_eq!(
from_compact, from_verbose,
"one decoder, two wires, identical RecordBatch"
);
}
/// Node and edge cells rebuild the exact verbose entity trees (ids resolved
/// through the catalogs), including the server's `dest_node` spelling.
#[test]
fn compact_node_and_edge_rebuild_verbose_trees() {
// As probed live: (a:A {k:1}), [r:R {w:1.5, tag:'t'}] — property ids
// under the `names()` fixture: k=4; w/tag are beyond it on purpose in
// the staleness test below, here we reuse k and s.
let compact = Value::Array(vec![
compact_header(&["a", "r"]),
Value::Array(vec![Value::Array(vec![
cell(
T_NODE,
Value::Array(vec![
Value::Int(1),
Value::Array(vec![Value::Int(1)]),
Value::Array(vec![Value::Array(vec![
Value::Int(4),
Value::Int(T_INTEGER),
Value::Int(1),
])]),
]),
),
cell(
T_EDGE,
Value::Array(vec![
Value::Int(0),
Value::Int(0),
Value::Int(1),
Value::Int(2),
Value::Array(vec![
Value::Array(vec![Value::Int(2), Value::Int(T_DOUBLE), bs("1.5")]),
Value::Array(vec![Value::Int(1), Value::Int(T_STRING), bs("t")]),
]),
]),
),
])]),
Value::Array(vec![]),
]);
let verbose = Value::Array(vec![
verbose_header(&["a", "r"]),
Value::Array(vec![Value::Array(vec![
Value::Array(vec![
Value::Array(vec![bs("id"), Value::Int(1)]),
Value::Array(vec![bs("labels"), Value::Array(vec![bs("A")])]),
Value::Array(vec![
bs("properties"),
Value::Array(vec![Value::Array(vec![bs("k"), Value::Int(1)])]),
]),
]),
Value::Array(vec![
Value::Array(vec![bs("id"), Value::Int(0)]),
Value::Array(vec![bs("type"), bs("R")]),
Value::Array(vec![bs("src_node"), Value::Int(1)]),
Value::Array(vec![bs("dest_node"), Value::Int(2)]),
Value::Array(vec![
bs("properties"),
Value::Array(vec![
Value::Array(vec![bs("f"), bs("1.5")]),
Value::Array(vec![bs("s"), bs("t")]),
]),
]),
]),
])]),
Value::Array(vec![]),
]);
let converted = compact_response_to_verbose(&compact, &names()).unwrap();
assert_eq!(converted, verbose);
// And both decode identically through the shared decoder.
assert_eq!(
response_to_batch_auto(converted).unwrap(),
response_to_batch_auto(verbose).unwrap()
);
}
/// An ordinal beyond the cached catalog is the refresh signal, never a
/// silent wrong name.
#[test]
fn unknown_ids_demand_catalog_refresh() {
let node = cell(
T_NODE,
Value::Array(vec![
Value::Int(0),
Value::Array(vec![Value::Int(99)]), // label id beyond catalog
Value::Array(vec![]),
]),
);
let resp = Value::Array(vec![
compact_header(&["n"]),
Value::Array(vec![Value::Array(vec![node])]),
Value::Array(vec![]),
]);
let err = compact_response_to_verbose(&resp, &names()).unwrap_err();
match err {
FlightSqlError::CompactIdUnknown { kind, id, have } => {
assert_eq!(kind, "label");
assert_eq!(id, 99);
assert_eq!(have, 3);
}
other => panic!("expected CompactIdUnknown, got {other:?}"),
}
}
/// A stats-only (write) reply is identical on both wires and passes through.
#[test]
fn stats_only_reply_passes_through() {
let resp = Value::Array(vec![Value::Array(vec![bs("Nodes created: 3")])]);
assert_eq!(
compact_response_to_verbose(&resp, &names()).unwrap(),
resp
);
}
/// A verbose reply fed to the compact converter is refused loudly — the
/// header entries are plain names, not `[type, name]` pairs.
#[test]
fn verbose_reply_is_refused_not_mangled() {
let resp = Value::Array(vec![
verbose_header(&["id"]),
Value::Array(vec![Value::Array(vec![Value::Int(1)])]),
Value::Array(vec![]),
]);
let err = compact_response_to_verbose(&resp, &names()).unwrap_err();
assert!(
matches!(&err, FlightSqlError::MalformedResponse(m) if m.contains("--compact")),
"got {err:?}"
);
}
}