nordisk/knut

graphar-flight — design

Class: design · Function axis: f0 the whole · f1 Flight SQL service · f2 query registry · f3 auth · f4 schema · f5 export. Reuse: flight.uses = [falkordb, graphar] — see falkordb, graphar. Strict: this file states only the shape we run (crates/graphar-flight/src). Every why, every rejected alternative, and the arrow-58 history live in flight-reasoning.md; the API/CLI workflow in flight-guide.md.

f0 — the whole

graphar-flight is an Arrow Flight SQL server that speaks Cypher in, Arrow out. A generic ADBC/ODBC driver — including Power BI's connector — connects, walks the Flight SQL metadata surface to discover catalogs / schemas / tables, then reads knut as if it were a SQL warehouse. The pipeline for one query:

1. Client sends Cypher directly, or SELECT * FROM <name> against a discovered table.
2. CypherFlightService::target_cypher (src/service.rs:162) resolves the string to real Cypher via the registry (f2).
3. The caller is authenticated by the interceptor and authorized per-query (f3).
4. FalkorExecutor (src/falkor.rs:15) runs GRAPH.QUERY against FalkorDB (falkordb).
5. convert::response_to_batch{,_auto} (src/convert.rs) turns the RESP reply into one Arrow RecordBatch; FlightDataEncoderBuilder streams it back (src/service.rs:243).

Two side surfaces hang off the same service: auth (one AuthConfig, f3) and export — export_cypher{,_chunked} and, under feature skade, straight to Iceberg (f5).

Laws (the contract this crate holds):

f1 — the Flight SQL service

pub struct CypherFlightService (src/service.rs:114) impl FlightSqlService (:254). Cloneable; holds a FalkorExecutor, an Arc<SchemaRegistry>, an Arc<AuthConfig>, and a shared SchemaCache. Built with connect(redis, graph, registry) (:125) then .with_auth(auth) (:145).

SchemaCache (src/service.rs:74): Arc<Mutex<HashMap<String, RecordBatch>>>, bounded by SCHEMA_CACHE_CAP = 16 (:63). insert evicts one entry when full for a new key (:82); take removes-and-returns (:95). It carries an auto-schema batch from GetFlightInfo to DoGet so the query is not run twice (L3). execute_cypher_stream (:229) branches: registered ⇒ typed exec.query; else cache hit ⇒ stream the parked batch; else ⇒ fresh query_auto. The static SQL_INFO (:40) advertises knut as read-only, Cypher-dialect, no SQL DDL/DML.

Server bootstrap (src/server.rs): run_server (:147) → run_server_with(.., ServerOptions) (:186). ServerOptions { tls, bearer_token, bearer_tokens, basic_auth, authorizer } (:22) folds to an AuthConfig via ServerOptions::auth_config (:49). The AuthInterceptor (:109) is installed only when auth.is_enabled() (:211); otherwise the server is open (transport-only / mTLS-only). TlsOptions (:66) carries server cert/key and, for mTLS, client_ca_pem.

f2 — the query registry

pub struct SchemaRegistry (src/registry.rs:57): HashMap<query, SchemaRef> plus a HashMap<table_name, query> name map. It is the source of L1 and L2.

parse_select_from(sql) (src/registry.rs:300) extracts the identifier from a trivial SELECT … FROM <ident> (stripping double-quote, backtick, or single-quote wrappers), the exact shape generic ADBC/ODBC drivers emit; target_cypher (service.rs:162) tries resolve_table(incoming), then resolve_table(parse_select_from(incoming)), else takes the string as Cypher verbatim.

f3 — auth (one config, two enforcement points)

src/auth.rs. pub struct AuthConfig (:162) holds bearer_tokens: Vec<(token, name)>, an optional BasicCredential (:41), and an Authorizer (:86).

• Authentication — authenticate(header) -> Identity (:288): open config ⇒ Identity::Anonymous; a matching Bearer (constant-time compared over every token, :300) ⇒ Identity::Token(name); valid Basic ⇒ Identity::Basic(user) (check_basic:316); else unauthenticated. check_header (:278) is the identity-discarding wrapper the handshake uses.
• Two points (L4): the per-call AuthInterceptor (server.rs:113) calls authenticate and stashes the Identity as a request extension; do_handshake (service.rs:269) calls the same check_header. The interceptor can't see the method name, so it accepts Bearer or Basic — which is what lets the Basic-bearing handshake pass through.
• Bearer set (L5): with_bearer_token / with_bearer_tokens / add_bearer_token (:179,:190,:219), plus with_named_bearer_tokens (:205) to map several tokens to one authz identity. issued_token (:269) is the first token; is_enabled (:263) gates the interceptor.
• Authorization (L6): Authorizer (:86) = AllowAll (default) | AllowList(HashMap<name, Set<query>>) with a "*" wildcard fallback (is_allowed:134) | Custom(Arc<dyn Fn>). The service's authorize<T> (service.rs:211) reads the Identity extension (absent ⇒ Anonymous), maps the resolved Cypher to its authz_name (:200), and calls AuthConfig::authorize (auth.rs:252); denial → permission_denied. Enforced in get_flight_info_statement, do_get_statement, both prepared read paths, and do_action_create_prepared_statement — every Cypher entry point.

constant_time_eq (auth.rs:330) backs both the token and the two-field Basic compare.

f4 — schema (auto-Utf8 vs typed)

Two paths, both ending in a RecordBatch:

• Auto-string (unregistered): FalkorExecutor::query_auto (falkor.rs:44) → convert::response_to_batch_auto — column names from FalkorDB's response header, every value Utf8. schema_for (service.rs:181) runs it once and parks the batch in SchemaCache (L3).
• Typed (registered): FalkorExecutor::query(cypher, schema) (falkor.rs:31) → convert::response_to_batch casts to the registered SchemaRef; _gar_id/src/dst come back Int64. Registered queries never cache and always execute in DoGet.

f5 — export (graph → lakehouse / Iceberg)

src/export.rs — the reverse bridge (falkordb writes Arrow in; this reads Cypher results out).

ExportReport { rows, columns, chunks } (export.rs:19). Chunked export bounds the write side, not the FalkorDB fetch — see flight-reasoning.md. Runnable: cargo run -p graphar-flight --example export (examples/export.rs) and, with --features skade, --example export-iceberg (examples/export-iceberg.rs).

DoPut: the parallel bulk path (2026-07)

Two things were wrong with do_put_statement_update, both fixed:

1. It dropped the schema. The handler did request.into_inner().into_inner(), and PeekableFlightDataStream::into_inner documents that "any messages observed through peek will be lost" — the SQL server peeks the first message to route the command, and every standard client (Arrow's own FlightDataEncoderBuilder, pyarrow's do_put(descriptor, schema)) puts the descriptor on the SAME message as the IPC schema. So the schema was eaten and the decoder failed with Received RecordBatch prior to Schema. It now uses into_peekable() and filters out any leading descriptor-only message (empty data_header), so both wire shapes work.

2. It was on the serial insert. It called insert_vertices / insert_edges, which await one UNWIND … CREATE at a time over the single link, while the in-process loader had been on insert_*_par (up to concurrency windows in flight over cloned multiplexed connections). Same crate, same renderer, two speeds. DoPut now calls the parallel pair, with the window count on ServerOptions::put_concurrency (default available_parallelism(); Some(1) pins it back to the serial dispatch).

What it buys, measured (NORNIR_MACHINE=oden, Threadripper PRO 3975WX, 32 physical cores, SMT off, 499GB RAM; FalkorDB on loopback; 40000 vertices):

speedup_x = 1.19, doput_vs_inprocess_x = 0.985.

Read that honestly: the Flight transport costs nothing (DoPut is within 1.5% of the in-process call), and the standing "DoPut is ~100x slower than in-process" note does not reproduce — routing was never the 100x. The ceiling here is FalkorDB itself: it serialises writes per graph at ≈28000 nodes/s, and a window sweep (512 → 40000 rows per UNWIND) moved it by under 10%. The client-side pipelining win is a round-trip win, so it materialises when FalkorDB is a network hop away — which is exactly what the DB-free knut.falkor_parallel_dispatch arm models (≈29.5x at a simulated 500us RTT), not on loopback.

Byte-identity is proven, not assumed: crates/graphar-flight/tests/integration_doput_parallel.rs runs the same batch through a serial-dispatch server, a parallel-dispatch server and the in-process loader, then dumps all three graphs (paged around FalkorDB's RESULTSET_SIZE cap) ordered by _gar_id with every property and asserts the dumps are equal.