crates/graphar-flight/tests/integration_compact.rs
//! **Live `--compact` wire parity** — the zero-loss pin for the compact query
//! door: the SAME live reply fetched over both wires must decode to EQUAL
//! `RecordBatch`es, and the lighter wire must actually be lighter (bench).
//!
//! Live tier — needs a FalkorDB. Repo convention (`#[ignore]`,
//! `$KNUT_FALKORDB_URL`, default `redis://127.0.0.1:6379`):
//!
//! ```bash
//! KNUT_FALKORDB_URL=redis://127.0.0.1:6399 \
//! cargo test -p graphar-flight --test integration_compact -- --ignored --nocapture
//! ```
use std::sync::Arc;
use std::time::Instant;
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use graphar_falkordb::FalkorDbLoader;
use graphar_flight::FalkorExecutor;
fn url() -> String {
std::env::var("KNUT_FALKORDB_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into())
}
/// A verbose executor and a compact executor on the same wiped graph.
async fn executors(graph: &str) -> (FalkorExecutor, FalkorExecutor) {
let mut verbose = FalkorExecutor::connect(&url(), graph)
.await
.unwrap_or_else(|e| panic!("connect to {} — {e}", url()));
let _ = verbose.execute("MATCH (n) DETACH DELETE n").await;
let compact = FalkorExecutor::connect(&url(), graph)
.await
.unwrap()
.with_compact(true);
(verbose, compact)
}
/// **The red-first seam: every decode-relevant value shape, both wires, one
/// assert per query.** Scalars (int / string / double incl. the integral-DOUBLE
/// `2.0` / boolean / null), collections, maps, points, nodes, edges, and an
/// array-valued node property — the verbose decode is the oracle and the
/// compact decode must equal it exactly.
#[tokio::test]
#[ignore = "needs live FalkorDB at $KNUT_FALKORDB_URL; run with --ignored"]
async fn compact_decode_equals_verbose_decode_live() {
let (mut verbose, mut compact) = executors("compact_parity").await;
verbose
.execute(
"CREATE (:C {i: 7, s: 'alpha', f: 0.5, g: 2.0, b: true, ls: [1, 'a', 2.5]}), \
(:C {i: -3, s: 'O\\'Brien \\\\ x', f: 1.5, g: 4.0, b: false, ls: []})",
)
.await
.expect("node fixture");
verbose
.execute(
"MATCH (a:C {i: 7}), (b:C {i: -3}) \
CREATE (a)-[:R {w: 1.5, tag: 't'}]->(b)",
)
.await
.expect("edge fixture");
let queries: &[&str] = &[
// Scalar columns, nulls (n.ghost) and the all-null Utf8 fallback.
"MATCH (n:C) RETURN n.i, n.s, n.f, n.g, n.b, n.ghost ORDER BY n.i",
// Expression scalars + collections + map + point.
"RETURN 1, 'x', 2.5, 2.0, true, null, [1, 'a', 2.0, null, [2]], \
{k: 1, s: 'x', d: 2.5, inner: [1, 'q']}, \
point({latitude: 57.7, longitude: 11.9})",
// Array-valued node property.
"MATCH (n:C) RETURN n.ls ORDER BY n.i",
// Whole graph entities: node and edge cells.
"MATCH (a:C)-[r:R]->(b:C) RETURN a, r, b",
// Zero rows (both wires must agree on the empty batch too).
"MATCH (n:C) WHERE n.i = 999999 RETURN n.i, n.s",
];
for q in queries {
let from_verbose = verbose.query_auto(q).await.unwrap_or_else(|e| panic!("verbose {q}: {e}"));
let from_compact = compact.query_auto(q).await.unwrap_or_else(|e| panic!("compact {q}: {e}"));
assert_eq!(
from_verbose, from_compact,
"compact decode must equal verbose decode for: {q}"
);
}
// The schema-driven door takes the same wire switch.
let schema: SchemaRef = Arc::new(Schema::new(vec![
Field::new("i", DataType::Int64, true),
Field::new("f", DataType::Float64, true),
Field::new("b", DataType::Boolean, true),
Field::new("s", DataType::Utf8, true),
]));
let q = "MATCH (n:C) RETURN n.i, n.f, n.b, n.s ORDER BY n.i";
let from_verbose = verbose.query(q, &schema).await.expect("verbose schema query");
let from_compact = compact.query(q, &schema).await.expect("compact schema query");
assert_eq!(from_verbose, from_compact, "schema-driven decode must agree");
// ro_query and execute (write stats) agree too.
let from_verbose = verbose.ro_query(q, &schema).await.expect("verbose ro_query");
let from_compact = compact.ro_query(q, &schema).await.expect("compact ro_query");
assert_eq!(from_verbose, from_compact, "ro_query decode must agree");
let nv = verbose.execute("CREATE (:W {x: 1}), (:W {x: 2})").await.expect("verbose write");
let nc = compact.execute("CREATE (:W {x: 3}), (:W {x: 4})").await.expect("compact write");
assert_eq!(nv, nc, "affected counts must agree across wires");
}
/// **Catalog staleness**: labels / property keys / relationship types minted
/// AFTER the compact executor warmed its caches must decode via the
/// refresh-on-miss path, not error and not mis-name.
#[tokio::test]
#[ignore = "needs live FalkorDB at $KNUT_FALKORDB_URL; run with --ignored"]
async fn compact_caches_refresh_on_miss() {
let (mut verbose, mut compact) = executors("compact_stale").await;
verbose.execute("CREATE (:Old {a: 1})").await.expect("seed");
// Warm the compact caches on the old world.
let warm = compact.query_auto("MATCH (n:Old) RETURN n").await.expect("warm");
assert_eq!(warm.num_rows(), 1);
// New label, new property keys, new relationship type — all unseen ids.
verbose
.execute("CREATE (:Fresh {brand_new: 42})-[:NEVER_SEEN {nu: 1}]->(:Fresh {brand_new: 7})")
.await
.expect("new world");
let q = "MATCH (a:Fresh)-[r:NEVER_SEEN]->(b:Fresh) RETURN a, r, b";
let from_verbose = verbose.query_auto(q).await.expect("verbose new world");
let from_compact = compact.query_auto(q).await.expect("compact new world (refresh-on-miss)");
assert_eq!(from_verbose, from_compact, "post-refresh decode must agree");
}
/// **Bench — the 100k-row results-back wire** (print-only; run with
/// `--nocapture`, and in `--release`: the parse/convert half is
/// allocation-shaped and a debug build inverts the verdict). Two query shapes,
/// medians of 5, equality asserted once at scale for each:
///
/// * `scalar` — `RETURN n.id, n.name, n._gar_id`: verbose rows carry NO
/// repeated names here, so compact's `[type, value]` pairs are strictly MORE
/// wire and more parse; expect compact ≥ verbose. (This is the shape the
/// 150–162 ms wire measurement of 2026-10-19 was taken on.)
/// * `entity` — `RETURN n`: verbose repeats `id`/`labels`/`properties`, every
/// label and every property NAME per row; compact sends ordinals. This is
/// the shape the flag is for.
#[tokio::test]
#[ignore = "needs live FalkorDB at $KNUT_FALKORDB_URL; bench — run with --ignored --nocapture"]
async fn bench_compact_wire_100k() {
const N: i64 = 100_000;
let graph = "compact_bench";
let mut loader = FalkorDbLoader::connect(&url(), graph).await.expect("loader");
let _ = loader.run_query("MATCH (n) DETACH DELETE n").await;
let names: Vec<String> = (0..N).map(|i| format!("n{i}")).collect();
let name_refs: Vec<&str> = names.iter().map(String::as_str).collect();
let nodes = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from((0..N).collect::<Vec<i64>>())),
Arc::new(StringArray::from(name_refs)),
],
)
.unwrap();
loader
.insert_vertices_par("P", &[nodes], 12)
.await
.expect("vertices");
let mut verbose = FalkorExecutor::connect(&url(), graph).await.expect("verbose exec");
let mut compact = FalkorExecutor::connect(&url(), graph)
.await
.expect("compact exec")
.with_compact(true);
fn median(mut xs: Vec<f64>) -> f64 {
xs.sort_by(|a, b| a.partial_cmp(b).unwrap());
xs[xs.len() / 2]
}
const REPS: usize = 5;
for (shape, q) in [
("scalar", "MATCH (n:P) RETURN n.id, n.name, n._gar_id LIMIT 100000"),
("entity", "MATCH (n:P) RETURN n LIMIT 100000"),
] {
// Equality at scale, once (also warms both paths + the compact catalogs).
let bv = verbose.query_auto(q).await.expect("verbose fetch");
let bc = compact.query_auto(q).await.expect("compact fetch");
assert_eq!(bv, bc, "100k-row decode must be identical across wires ({shape})");
assert_eq!(bv.num_rows(), N as usize);
let mut wall_v = Vec::new();
let mut wall_c = Vec::new();
for _ in 0..REPS {
let t = Instant::now();
let b = verbose.query_auto(q).await.expect("verbose rep");
wall_v.push(t.elapsed().as_secs_f64() * 1000.0);
assert_eq!(b.num_rows(), N as usize);
let t = Instant::now();
let b = compact.query_auto(q).await.expect("compact rep");
wall_c.push(t.elapsed().as_secs_f64() * 1000.0);
assert_eq!(b.num_rows(), N as usize);
}
println!(
"BENCH compact_wire_100k[{shape}]: verbose wall median {:.1} ms | compact wall median {:.1} ms | saved {:.1} ms ({:.2}x)",
median(wall_v.clone()),
median(wall_c.clone()),
median(wall_v.clone()) - median(wall_c.clone()),
median(wall_v) / median(wall_c),
);
}
}