nordisk/knut

knut — complete overview

The real README: the full pitch, quickstart, design, examples and the measured test/bench matrix. The repo-root README.md is the slim face (title · mashup · pointers); this chapter is the heavy detail, and it is also folded into the manual docs/manual.pdf by nornir docs book.

knut

A pure-Rust columnar ↔ graph fabric: read/write Apache GraphAr files, bulk-load them into FalkorDB, and serve Cypher over Arrow Flight SQL — plus Bifröst, which streams Iceberg tables into a live graph by CDC (Spark-Connect or Spark-free). One Arrow pipe from the lakehouse to a queryable graph.

Why knut — it carries graph data the whole way from a columnar lake to a live, queryable graph in one Arrow pipe (no JVM required) — the rows marked NA are categories the rivals have no concept for:

Full numbers — generated by nornir bench run knut. Rows marked NA: the rival has no such category.

New here? → knut for dummies — a hands-on, copy-paste tour that starts the containers and runs every feature.

Design docs. This README is the face. The deep design lives in the nornir doc set under .nornir/: the map and one territory doc per subsystem — graphar · graphar-falkordb · graphar-flight · knut-bifrost · knut-pipelines. Locked decisions and open work are in .nornir/plan.md; rendered docs (PDF + HTML) in docs/.

What it does

• graphar — read & write Apache GraphAr (gar/v1) property graphs as chunked columnar files: 5 formats (Parquet/CSV/ORC/JSON/Arrow IPC), local filesystem or any object store (S3/GCS/Azure/SeaweedFS/RustFS).
• graphar-falkordb — bulk-load Arrow RecordBatches into a FalkorDB graph via batched UNWIND … CREATE Cypher, endpoints matched by the shared _gar_id linking key.
• graphar-flight — an Arrow Flight SQL gRPC server that takes Cypher in and streams Arrow out; auth (bearer / Basic-handshake / mTLS); graph → lakehouse export (GraphAr or Iceberg). Any ADBC / Power BI / DataFusion client works.
• knut-bifrost (Bifröst 🌈) — watch Iceberg tables (Spark Connect or Spark-free via skade) and continuously materialise them into FalkorDB with changelog CDC and an idempotent MERGE sink.
• knut (Python) + knut-pipelines — a pure-Python Flight SQL client (Cypher → pandas → NetworkX) and a live viewer for Spark Declarative Pipelines / Connect jobs (knutpipe).

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        GraphAr files (disk)                     │
│   vertex/Person/id/chunk0.parquet                               │
│   edge/Person_knows_Person/ordered_by_source/adj_list/...       │
└───────────────┬─────────────────────────────────────────────────┘
                │ graphar (Rust)
                │ VertexReader / EdgeReader / VerticesBuilder / EdgesBuilder
                ▼
┌─────────────────────────────────────────┐
│   Apache Arrow RecordBatch (in-memory)  │
└───────────────┬─────────────────────────┘
                │ graphar-falkordb (Rust)
                │ FalkorDbLoader — UNWIND … CREATE (Cypher)
                ▼
┌─────────────────────────────────────────┐
│   FalkorDB  (Redis / Valkey module)     │
│   nodes with _gar_id property           │
└───────────────┬─────────────────────────┘
                │ graphar-flight (Rust)
                │ Arrow Flight SQL gRPC — GRAPH.QUERY → RecordBatch stream
                ▼
┌─────────────────────────────────────────┐
│   Arrow Flight SQL wire protocol        │
└───────────────┬─────────────────────────┘
                │ knut (Python)
                │ KnutClient — DoGet / DoPut
                ▼
┌─────────────────────────────────────────┐
│   pandas DataFrame  →  NetworkX Graph   │
│   nx.from_pandas_edgelist               │
│   nx.set_node_attributes                │
└─────────────────────────────────────────┘

Components

Rust crates

Python client

Prerequisites

Start FalkorDB with Docker:

podman run -p 6379:6379 localhost/knut-falkordb-valkey9:latest

Quick start

1. Write a graph to disk (GraphAr format)

use graphar::{
    VertexInfo, PropertyGroup, Property, DataType, FileType,
    VerticesBuilder, PropertyValue, Vertex,
};

let vi = VertexInfo::new("Person", 262_144)
    .with_property_group(PropertyGroup::new(
        vec![
            Property::new("id", DataType::Int64).primary(),
            Property::new("name", DataType::String),
        ],
        FileType::Parquet,
    ));

let mut builder = VerticesBuilder::new(vi, "./graph");
builder.add_vertex(Vertex::new()
    .add_property("id",   PropertyValue::Int64(0))
    .add_property("name", PropertyValue::String("Alice".into())));
builder.dump()?;

2. Load into FalkorDB

use graphar_falkordb::FalkorDbLoader;

let mut loader = FalkorDbLoader::connect("redis://127.0.0.1:6379", "mygraph").await?;
loader.create_index("Person", "_gar_id").await?;
loader.insert_vertices("Person", &batches).await?;
loader.insert_edges("Person", "KNOWS", "Person", &edge_batches).await?;

3. Start the Flight SQL server

use std::sync::Arc;
use graphar_flight::{SchemaRegistry, run_server};

// Empty registry works — types are inferred automatically from FalkorDB's response.
// Optionally register queries for explicit typed schemas.
let registry = Arc::new(SchemaRegistry::new());

registry.register_vertex_query(
    "MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name",
    &["name"],
);
registry.register_edge_query(
    "MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN a._gar_id AS src, b._gar_id AS dst",
    &[],
);

run_server("redis://127.0.0.1:6379", "mygraph", registry, "0.0.0.0:50051".parse()?).await?;

4. Query from Python → NetworkX

from knut import KnutClient

G = KnutClient().to_networkx(
    vertex_query="MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name",
    edge_query="MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN a._gar_id AS src, b._gar_id AS dst",
)
print(G.nodes[0])  # {'name': 'Alice'}

The _gar_id linking key

All components share a single identifier: _gar_id, the 0-based row index of each vertex across all batches. It is the glue between GraphAr's internal vertex ID, FalkorDB node properties, and NetworkX node IDs.

GraphAr row index 0  ──►  FalkorDB node { _gar_id: 0, name: "Alice" }
                                │
            edge { src: 0, dst: 1 }  ──►  MATCH (a { _gar_id: 0 }), (b { _gar_id: 1 })
                                │
                     NetworkX node 0  with attrs { name: "Alice" }

• graphar-falkordb sets _gar_id on every vertex during insert_vertices.
• insert_edges matches endpoints by _gar_id, so no user-defined primary key is required.
• knut uses _gar_id as the NetworkX node identifier and join key for node attributes.

graphar

Implements the GraphAr gar/v1 specification for reading and writing property graphs as chunked columnar files.

Concepts

• Vertex chunks — a vertex type is split into fixed-size chunks (default 2¹⁸ = 262,144 rows), each chunk stored as a separate Parquet or CSV file.
• Property groups — properties are stored in separate column groups so only the columns you need are read from disk.
• Edge chunks — edges are partitioned by vertex chunk alignment (CSR / CSC / COO), with offset tables for ordered adjacency lists enabling O(degree) edge lookups.
• Metadata — GraphInfo, VertexInfo, and EdgeInfo YAML files describe the full schema and file layout. They are the entry point for both readers and writers.

File layout

<prefix>/
  <graph>.graph.yml
  <label>.vertex.yml
  vertex/<label>/<prop_group>/chunk0.parquet
  <src>_<edge>_<dst>.edge.yml
  edge/<src>_<edge>_<dst>/ordered_by_source/adj_list/part0/chunk0.parquet
  edge/<src>_<edge>_<dst>/ordered_by_source/offset/chunk0.parquet
  edge/<src>_<edge>_<dst>/ordered_by_source/<prop_group>/part0/chunk0.parquet

part{n} is the vertex-chunk partition index; chunk{n} is the edge-chunk index within that partition.

Writing a graph

use graphar::{
    AdjListInfo, AdjListType, DataType, Edge, EdgeInfo,
    EdgesBuilder, FileType, Property, PropertyGroup, PropertyValue,
    Vertex, VertexInfo, VerticesBuilder,
};

// Define vertex schema
let vertex_info = VertexInfo::new("Person", 100)
    .with_property_group(PropertyGroup::new(
        vec![
            Property::new("id", DataType::Int64).primary(),
            Property::new("name", DataType::String),
        ],
        FileType::Parquet,
    ));

// Write vertices
let mut vb = VerticesBuilder::new(vertex_info.clone(), "./data");
vb.add_vertex(Vertex::new()
    .add_property("id",   PropertyValue::Int64(0))
    .add_property("name", PropertyValue::String("Alice".into())));
vb.dump()?;

// Define edge schema (CSR — ordered by source vertex)
let edge_info = EdgeInfo::new("Person", "knows", "Person", 1024, 100, 100, false)
    .with_adj_list(AdjListInfo {
        ordered: true,
        aligned_by: "src".into(),
        file_type: FileType::Parquet,
        prefix: None,
    });

// Write edges
let mut eb = EdgesBuilder::new(edge_info, "./data", AdjListType::OrderedBySource, 2);
eb.add_edge(Edge::new(0, 1));
eb.dump()?;

Reading a graph

use graphar::{GraphInfo, VertexReader};

let graph_info = GraphInfo::load("./data/my_graph.graph.yml")?;
let vertex_info = graph_info.get_vertex_info("Person", "./data")?;

let reader = VertexReader::new(vertex_info.clone(), "./data", 1000);
let group = &vertex_info.property_groups[0];
let batches = reader.read_property_group(group)?;   // Vec<RecordBatch>

for batch in &batches {
    println!("{} rows", batch.num_rows());
}

Supported file formats

All five formats work both on the local filesystem (graphar::io) and against any object store (graphar::storage::ChunkStore): S3, GCS, Azure, plus S3-compatible endpoints — SeaweedFS, RustFS, MinIO — via ChunkStore::s3_compatible(endpoint, bucket, key, secret, prefix).

Supported data types

Bool, Int32, Int64, Float, Double, String, Date, Timestamp, Time, List

graphar-falkordb

Bulk-loads Apache Arrow RecordBatches into a FalkorDB graph (Redis/Valkey) using batched UNWIND … CREATE Cypher queries.

How it works

Each GRAPH.QUERY call sends up to 512 rows (configurable) as an inline UNWIND list:

UNWIND [{`_gar_id`: 0, `name`: 'Alice'}, {`_gar_id`: 1, `name`: 'Bob'}] AS row
CREATE (n:`Person`) SET n = row

For edges, the server MATCHes the endpoints by _gar_id and sets edge properties:

UNWIND [{`_src`: 0, `_dst`: 1, `since`: '2020'}] AS row
MATCH (a:`Person` {`_gar_id`: row._src}), (b:`Person` {`_gar_id`: row._dst})
CREATE (a)-[r:`KNOWS`]->(b) SET r = row
REMOVE r._src, r._dst

Usage

use graphar_falkordb::FalkorDbLoader;

let mut loader = FalkorDbLoader::connect("redis://127.0.0.1:6379", "mygraph").await?;

// Index _gar_id before inserting edges — makes MATCH fast
loader.create_index("Person", "_gar_id").await?;

// Insert vertices — each RecordBatch column becomes a node property
loader.insert_vertices("Person", &vertex_batches).await?;

// Insert edges — batch must have Int64 `src` and `dst` columns
loader.insert_edges("Person", "KNOWS", "Person", &edge_batches).await?;

Merging GraphAr property groups before insert

GraphAr stores properties split across multiple group files. Merge them horizontally before inserting into FalkorDB:

VertexReader  →  [PropertyGroup RecordBatch, ...]
                        ↓
           merge_property_groups(groups)        — or reader.read_merged()
                        ↓
     FalkorDbLoader::insert_vertices(label, &[merged_batch])

Batch size

Default is 512 rows per GRAPH.QUERY call. Tune with .with_batch_size(n)?.

Arrow → Cypher type mapping

graphar-flight

An Apache Arrow Flight SQL gRPC server that accepts Cypher queries instead of SQL, executes them against FalkorDB via GRAPH.QUERY, and streams results as Arrow RecordBatches.

Any Flight SQL-compatible client works: ADBC, DataFusion, DBeaver (Flight plugin), flightsql-dbapi (Python), the knut Python client included in this repo, etc.

Protocol flow

Client                              Server (graphar-flight)
  │                                        │
  │── GetFlightInfo(CommandStatementQuery) ─►│ look up schema in registry
  │◄─ FlightInfo(schema, ticket) ───────────│
  │                                        │
  │── DoGet(ticket) ───────────────────────►│ GRAPH.QUERY → RecordBatch
  │◄─ FlightData stream ────────────────────│
  │                                        │
  │── DoPut(CommandStatementUpdate,        │
  │         Arrow stream) ────────────────►│ UNWIND … CREATE in FalkorDB
  │◄─ affected row count ───────────────────│

Schema registry (optional)

The server has two modes:

In auto mode the server peeks at the first non-null value in each column to determine its Arrow type, so integer IDs like _gar_id, src, and dst arrive as Int64 by default — correct NetworkX node IDs with no configuration needed.

Registering typed schemas (optional, but faster)

Two convenience helpers cover the common patterns. Both set _gar_id, src, and dst to Int64 (native integers, not strings), and append any extra column names as Utf8.

use std::sync::Arc;
use graphar_flight::{SchemaRegistry, run_server};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // An empty registry is valid — unregistered queries use auto-string mode.
    let registry = Arc::new(SchemaRegistry::new());

    // Vertex query: _gar_id → Int64, extra columns → Utf8.
    // The RETURN clause must alias the internal ID as _gar_id and list
    // extra columns in the same order as the slice.
    registry.register_vertex_query(
        "MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name, n.age",
        &["name", "age"],
    );

    // Edge query: src + dst → Int64, extra columns → Utf8.
    // Alias source _gar_id as src, destination _gar_id as dst.
    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"],
    );

    // Edges with no properties — just topology.
    registry.register_edge_query(
        "MATCH (a:Person)-[:FOLLOWS]->(b:Person) \
         RETURN a._gar_id AS src, b._gar_id AS dst",
        &[],
    );

    // Full control: use register() directly for custom types (e.g. age as Int64).
    // registry.register("...", Arc::new(Schema::new(vec![...])));

    run_server(
        "redis://127.0.0.1:6379",
        "mygraph",
        registry,
        "0.0.0.0:50051".parse()?,
    ).await?;

    Ok(())
}

Bulk insert (DoPut)

The CommandStatementUpdate query string is a directive, not a Cypher statement:

Redis → Arrow type mapping

Both registered and auto-string modes use the same coercions. In auto-string mode the type is inferred from the first non-null value in each column.

Integer properties like _gar_id, src, and dst arrive as Int64 in both modes — no registration needed for correct NetworkX node IDs.

Python client (knut)

Located in python/. A pure-Python Arrow Flight SQL client that queries the graphar-flight server and builds NetworkX graphs directly from the Arrow results.

No intermediate format conversion is needed: Arrow → pandas → NetworkX uses nx.from_pandas_edgelist and nx.set_node_attributes directly.

Install

cd python
pip install -e ".[dev]"   # installs pyarrow, pandas, networkx

Query → NetworkX

No schema pre-registration is required. The server infers Arrow types from FalkorDB's response — integers arrive as Int64, floats as Float64, strings as Utf8. NetworkX node IDs are integers by default.

from knut import KnutClient

client = KnutClient(host="localhost", port=50051)

# One call: fetch vertices + edges, return a NetworkX DiGraph
G = client.to_networkx(
    vertex_query="MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name, n.age",
    edge_query=(
        "MATCH (a:Person)-[r:KNOWS]->(b:Person) "
        "RETURN a._gar_id AS src, b._gar_id AS dst, r.since"
    ),
)

print(G.number_of_nodes(), G.number_of_edges())
print(G.nodes[0])   # {'name': 'Alice', 'age': 30}
print(G[0][1])      # {'since': '2020-01-01'}

# Standard NetworkX algorithms work on the result
print(nx.in_degree_centrality(G))
print(nx.shortest_path(G, source=0, target=3))

Insert

import pandas as pd
from knut import KnutClient

client = KnutClient()

# Vertices — _gar_id is added automatically (0-based row index)
client.insert_vertices("Person", pd.DataFrame({
    "name": ["Alice", "Bob", "Carol"],
    "age":  [30, 25, 35],
}))

# Edges — src/dst reference _gar_id values of the endpoint nodes
client.insert_edges("Person", "KNOWS", "Person", pd.DataFrame({
    "src":   [0, 0, 1],
    "dst":   [1, 2, 2],
    "since": ["2020-01-01", "2019-06-15", "2021-03-10"],
}))

Raw Arrow query

import pyarrow as pa

table: pa.Table = client.query("MATCH (n:Person) RETURN n._gar_id AS _gar_id, n.name")
df = table.to_pandas()

Proto encoding

The client speaks raw gRPC Flight SQL without any generated proto files or grpc_tools.
knut/_proto.py hand-encodes the two Flight SQL messages used
(CommandStatementQuery, CommandStatementUpdate) with a small varint
wire-format encoder — the only dependency is pyarrow.

API reference

Tests

cd python
pytest tests/ -v   # 13 tests, no server required (Rust: 19 unit + 6 doctests)

Building & testing

# Rust — build and test all crates
cargo build
cargo test

# Python — install and test
cd python
pip install -e ".[dev]"
pytest tests/ -v

Running the full stack locally

# 1. Start FalkorDB
podman run -p 6379:6379 localhost/knut-falkordb-valkey9:latest

# 2. Build and run the Flight SQL server (see graphar-flight example above)
cargo run --example server

# 3. Run the Python demo
cd python
python examples/social_graph.py

Bifröst 🌈 — Iceberg tables → FalkorDB graph

knut-bifrost watches Iceberg 1.11+ tables and continuously materialises them as a property graph: snapshot polling, incremental reads, a declarative-or-custom transform fabric, and an idempotent MERGE sink.

Two read lanes: Spark-free (IcebergSource, feature skade — pure Rust, no JVM, a local warehouse directory) and Spark Connect (SparkConnectSource, feature spark — the only route to a remote REST/Glue/Hive catalog or an S3-backed warehouse). Both are library-only; the bifrost CLI wires FileSource only. Which one to pick, what must be running for each, and the transfer-method menu into FalkorDB: iceberg-falkordb-guide.md. Full design: bifrost-design.md.

# bifrost.yml
version: 1
tables:
  - table: lake.social.users
    mode: incremental
    pipeline:
      - vertex: { label: Person, id_column: user_id, properties: [name] }
      - fk_edges:
          type: WORKS_AT
          src: { label: Person,  column: user_id }
          dst: { label: Company, column: company_id }

let source = SparkConnectSource::connect("sc://spark:15002").await?; // feature "spark"
let sink   = FalkorDbSink::connect("redis://127.0.0.1:6379", "social").await?;
let mut l  = Listener::new(MappingSpec::load("bifrost.yml")?, TransformRegistry::new(),
                           source, sink, SyncState::open("state.json")?);
l.run(Duration::from_secs(30), |round| { println!("{round:?}"); true }).await?;

Test matrix

Unit tests (cargo test --workspace, no containers) and a podman-backed integration matrix (-- --ignored, via testcontainers over the podman socket). Container images for the matrix live in containers/.

# build the matrix images once
podman build -t knut-falkordb-valkey9 -f containers/falkordb-valkey9/Containerfile .
podman build -t knut-spark-iceberg    -f containers/spark-iceberg/Containerfile .

# run everything that needs podman
cargo test --workspace -- --ignored
cargo test -p knut-bifrost --features spark --test integration_spark_falkordb -- --ignored

Benchmarks

knut follows the nornir bench contract — one suite, single-shot real workloads, one BenchRun JSON line for the warehouse. See .nornir/benching.md.

cargo run -p xtask --release --example nornir-bench   # canonical
cargo bench -p xtask                                  # same suite

Roadmap

The full proposition list with rationale: .nornir/plan.md.

License

Apache-2.0