nordisk/knut

graphar-flight — guide

Class: guide · audience: a techie standing up or driving the Cypher Flight SQL server. Reuse: flight.uses = [falkordb, graphar] — see falkordb, graphar. The what is in flight-design.md; the why in flight-reasoning.md. This file is the workflow: Rust API + CLI.

f0 — the whole, in four moves

1. Build a registry, naming the queries you want visible as tables (register_named).
2. Start the server — run_server (open) or run_server_with(ServerOptions) for TLS/auth.
3. Point a client at it — any ADBC/ODBC Flight SQL driver / Power BI handshakes (Basic → bearer), walks the metadata, runs SELECT * FROM <name>.
4. Export results back to the lakehouse with export_cypher{,_chunked}.

f1 — register queries (the tables)

use std::sync::Arc;
use graphar_flight::SchemaRegistry;
use arrow_schema::{DataType, Field, Schema};

let registry = Arc::new(SchemaRegistry::new());

// Friendly table name → Cypher + schema. `SELECT * FROM people` resolves here.
registry.register_named("people",
    "MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name",
    Arc::new(Schema::new(vec![
        Field::new("_gar_id", DataType::Int64, false),
        Field::new("name",    DataType::Utf8,  true),
    ])));

// Typed helpers — _gar_id / src / dst come back Int64, extras Utf8.
registry.register_vertex_query(
    "MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name, n.city", &["name", "city"]);
registry.register_edge_query(
    "MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN a._gar_id AS src, b._gar_id AS dst, r.since",
    &["since"]);

An empty registry is valid — every query then runs in auto-Utf8 mode. register_named (registry.rs:216), register_vertex_query (:130), register_edge_query (:177).

f2 — run the server

// Open: no auth, no TLS.
use graphar_flight::run_server;
run_server("redis://127.0.0.1:6379", "mygraph", registry, "0.0.0.0:50051".parse()?).await?;

With TLS + auth — run_server_with(.., ServerOptions { .. }) (server.rs:186); all methods compose:

use graphar_flight::{run_server_with, ServerOptions, TlsOptions};
let opts = ServerOptions {
    // mTLS (transport identity): server cert/key + the CA that verifies client certs.
    tls: Some(TlsOptions::mutual_from_pem_files("server.crt", "server.key", "client-ca.crt")?),
    basic_auth: Some(("analyst".into(), "p@ss".into())), // handshake trades this for a bearer token
    bearer_token: Some("s3cret".into()),                 // the token the handshake issues
    ..Default::default()
};
run_server_with("redis://127.0.0.1:6379", "mygraph", registry, "0.0.0.0:50051".parse()?, opts).await?;

TlsOptions::from_pem_files (server.rs:78) is server-only TLS; mutual_from_pem_files (:91) requires a client cert.

Bearer-token rotation

bearer_token is the issued token; bearer_tokens adds the rest of the accepted set. Any listed token authenticates, so rotate by add new → migrate clients → drop old (ServerOptions:38):

let opts = ServerOptions {
    bearer_token: Some("new-tok".into()),      // issued to fresh handshakes
    bearer_tokens: vec!["old-tok".into()],     // still accepted during the overlap
    ..Default::default()
};   // "Bearer new-tok" and "Bearer old-tok" pass; "Bearer retired" is rejected.

Per-query authorization

Authorizer::allow_list keys on the identity name (bearer-token name or Basic username) and permits only the listed table names; deny ⇒ permission_denied (auth.rs:Authorizer:86). Map several tokens to one identity with with_named_bearer_tokens (auth.rs:205):

use graphar_flight::{AuthConfig, Authorizer};
let auth = AuthConfig::default()
    .with_named_bearer_tokens([("tok-old", "analyst"), ("tok-new", "analyst")])
    .with_authorizer(Authorizer::allow_list([("analyst", vec!["people"])]));
// analyst → people: Ok;  analyst → salaries: permission_denied.

Wire it through ServerOptions { authorizer: Authorizer::allow_list([("analyst", vec!["people"])]), .. } (server.rs:45). Authorizer::custom(|id, q| …) (auth.rs:126) is an arbitrary closure; a "*" key is the wildcard fallback for identities not otherwise listed.

f3 — client-side: metadata + query + insert

A generic Flight SQL client (Power BI's ADBC connector, adbc_driver_flightsql, the stock FlightSqlServiceClient) needs no knut-specific code: Basic handshake → bearer, then GetCatalogs/GetDbSchemas/GetTables/GetTableTypes/GetSqlInfo, then SELECT * FROM people. Prepared statements work too (handle = resolved Cypher); the end-to-end Power BI path is exercised in tests/integration_flight_sql.rs (--ignored, needs a live FalkorDB).

Insert via DoPut — set the CommandStatementUpdate query to a directive and stream Arrow (service.rs:672): VERTICES:<label> bulk-inserts vertices, EDGES:<src>:<type>:<dst> edges.

f4 — export back to the lakehouse

use graphar_flight::{FalkorExecutor, export::{export_cypher, export_cypher_chunked}};
use graphar::FileType;

let mut exec = FalkorExecutor::connect("redis://127.0.0.1:6379", "social").await?;
let q = "MATCH (n:Person) RETURN n._gar_id AS id, n.name AS name";

// One file (format from the FileType).
let r = export_cypher(&mut exec, q, "/tmp/people.parquet", FileType::Parquet).await?;
// Chunked: chunk0.parquet, chunk1.parquet, … each ≤ chunk_rows; r.chunks lists them.
let r = export_cypher_chunked(&mut exec, q, "/tmp/people_chunks", FileType::Parquet, 100_000).await?;

export_cypher (export.rs:54), export_cypher_chunked (:122) → ExportReport { rows, columns, chunks } (:19). Chunked bounds write-side memory, not the FalkorDB fetch — see flight-reasoning.md.

CLI (format inferred from the output extension: parquet/csv/json/orc/arrow):

cargo run -p graphar-flight --example export -- \
  redis://127.0.0.1:6379 mygraph \
  "MATCH (n:Person) RETURN n._gar_id AS id, n.name AS name" /tmp/people.parquet

Straight to Iceberg (feature skade) — export_cypher_to_iceberg(&mut exec, q, warehouse, table) (export.rs:218) appends into an embedded Iceberg warehouse (no container; catalog is redb in the given dir):

cargo run -p graphar-flight --features skade --example export-iceberg -- \
  redis://127.0.0.1:6379 mygraph \
  "MATCH (n:Person) RETURN n._gar_id AS id, n.name AS name" /tmp/knut_warehouse people