# 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](falkordb-design.md), [graphar](graphar-design.md).
> 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`](flight-reasoning.md);
> the API/CLI workflow in [`flight-guide.md`](flight-guide.md).
## f0 — the whole
`graphar-flight` is an [Arrow Flight SQL](https://arrow.apache.org/docs/format/FlightSql.html)
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--the-query-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):**
| law | statement | source |
|---|---|---|
| **L1** registered-queries-are-tables | a registered query is advertised in `GetTables`; `SELECT * FROM <name>` is rewritten to its Cypher | `registry.rs:register_named`, `parse_select_from`, `service.rs:target_cypher` |
| **L2** auto-schema by default, typed on request | unregistered ⇒ all-`Utf8` inferred; a registered vertex/edge query promotes `_gar_id`/`src`/`dst` to `Int64` | `registry.rs:register_vertex_query`/`register_edge_query`, `falkor.rs:query_auto` |
| **L3** inferred schema executes **once** (H9) | auto-schema `GetFlightInfo` parks its batch in `SchemaCache`; the matching `DoGet` removes-and-streams it | `service.rs:SchemaCache`, `schema_for`, `execute_cypher_stream` |
| **L4** one `AuthConfig`, two enforcement points | the same config drives the per-call interceptor **and** `do_handshake`; interceptor accepts Bearer **or** Basic | `server.rs:AuthInterceptor`, `service.rs:do_handshake`, `auth.rs:authenticate` |
| **L5** bearer tokens are a **set** | a list of accepted tokens; any authenticates, the first is issued — zero-downtime rotation | `auth.rs:bearer_tokens`, `issued_token` |
| **L6** authenticate → `Identity`, then authorize | interceptor resolves an `Identity`; the service consults `Authorizer` on the resolved query name at every Cypher entry point | `auth.rs:authenticate`/`Authorizer`, `service.rs:authorize` |
| **L7** prepared statements are stateless | the handle **is** the resolved Cypher — no server-side statement table, any node serves any handle | `service.rs:do_action_create_prepared_statement`, `do_get_prepared_statement` |
## 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`).
| handler | role | source |
|---|---|---|
| `do_handshake` | Basic login → issues a bearer token in both payload and the `authorization` trailer; reuses `AuthConfig::check_header` | `service.rs:269` |
| `get_flight_info_statement` | resolve Cypher, authorize, describe schema (`schema_for`), return a ticket carrying the **resolved** Cypher | `service.rs:307` |
| `do_get_statement` | decode the ticket, authorize, `execute_cypher_stream` | `service.rs:355` |
| `do_put_statement_update` | `VERTICES:`/`EDGES:` directive → `FalkorDbLoader::insert_vertices_par`/`insert_edges_par` (the **parallel** bulk path, `put_concurrency` windows in flight) | `service.rs`, `parse_put_directive` |
| metadata: `get_flight_info_*` / `do_get_*` for catalogs, schemas, tables, table-types, sql-info | one catalog (the graph), one schema (`public` = `DB_SCHEMA:618`), one "table" per registered query | `service.rs:425`–`537` |
| prepared statements | `do_action_create_prepared_statement` / `get_flight_info_prepared_statement` / `do_get_prepared_statement` — handle is the Cypher (L7) | `service.rs:543`,`570`,`595` |
`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.
| method | role | source |
|---|---|---|
| `register` | store an arbitrary schema under a Cypher key | `registry.rs:96` |
| `register_vertex_query(query, extra_cols)` | `_gar_id: Int64` (non-null) + each extra as nullable `Utf8` | `registry.rs:130` |
| `register_edge_query(query, extra_cols)` | `src: Int64`, `dst: Int64` (non-null) + extras as `Utf8` | `registry.rs:177` |
| `register_named(name, query, schema)` | store the schema **and** alias `name → query` for `GetTables` | `registry.rs:216` |
| `resolve_table(ident)` | name alias wins, else a registered Cypher key, else `None` | `registry.rs:259` |
| `name_of(query)` | inverse of the alias — the table name for authz (`Authorizer` keys on it) | `registry.rs:247` |
| `tables()` | `(table_name, schema)` per registered query for `GetTables` | `registry.rs:272` |
| `get(key)` | schema by exact key; `None` ⇒ auto-schema | `registry.rs:232` |
`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).
| fn | role | source |
|---|---|---|
| `export_cypher(exec, cypher, path, FileType)` | `query_auto` → one GraphAr chunk file (`graphar::io::write_chunk`) | `export.rs:54` |
| `export_cypher_chunked(exec, cypher, dir, FileType, chunk_rows)` | fetch, then `write_batch_chunked` | `export.rs:122` |
| `write_batch_chunked(batch, dir, ft, chunk_rows)` | zero-copy `RecordBatch::slice` into `chunk{i}.<ext>`; `ExportReport.chunks` lists them; empty ⇒ one empty `chunk0` | `export.rs:141` |
| `export_cypher_to_iceberg(exec, cypher, warehouse, table)` (feat `skade`) | `query_auto` → `append_arrow_to_iceberg` | `export.rs:218` |
| `append_arrow_to_iceberg(dir, table, batch)` (feat `skade`) | `bridge_56_to_57` → `skade::Warehouse::table_or_create.append` | `export.rs:242` |
| `bridge_56_to_57(&batch)` (feat `skade`) | **identity clone** — knut + skade share Arrow 58, so this is a plain `clone()` (Arc bump, no copy) | `export.rs:327` |
`ExportReport { rows, columns, chunks }` (`export.rs:19`). Chunked export bounds the **write** side,
not the FalkorDB fetch — see [`flight-reasoning.md`](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):
| arm | rows/s |
|---|--:|
| serial dispatch (`put_concurrency = 1`) | 23303 |
| parallel dispatch (`put_concurrency = 32`) | 27712 |
| in-process `insert_vertices_par` (no Flight wire) | 27307 |
`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.