nordisk/knut

.nornir/README-full.md

# knut — complete overview

> The **real** README: the full pitch, quickstart, design, examples and the
> measured test/bench matrix. The repo-root [`README.md`](../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`](../docs/manual.pdf) by `nornir docs book`.

<p align="center"><img src="assets/generated_knut_iceberg.png" alt="knut — a knytpunkt over graphs" width="560"></p>

# 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:

| capability | **knut** | GraphAr (official) | Neo4j / GDS | Spark + GraphFrames |
|---|:--:|:--:|:--:|:--:|
| Pure-**Rust** GraphAr — 5 formats (Parquet/CSV/ORC/JSON/Arrow), local **or** object store | **✅** | ◐ (C++/Java/Py) | ✗ | ◐ |
| **Cypher over Arrow Flight SQL** server — ADBC / Power BI / DataFusion clients, Arrow out | **✅** | **NA** | ◐ | ✗ |
| **Iceberg → graph CDC** fabric — snapshot-incremental + changelog (`_change_type` → MERGE / DETACH DELETE) | **✅** | **NA** | ✗ | ◐ |
| **Dual-engine CDC** — Spark 4.2's native `CHANGES` clause *and* Iceberg's `create_changelog_view`, chosen by **probing** the live server | **✅** | **NA** | ✗ | ◐ |
| CDC **fallback is never silent** — `fell_back` + the engine's own reason in the log **and** in `state_json` | **✅** | **NA** | ✗ | ✗ |
| **Renderable** change rows — `(op, before, after, snapshot, ts)` Arrow, streamed batch-by-batch to a UI pane | **✅** | **NA** | ◐ | ✗ |
| **Projection + predicate pushed into the CDC read** (not just full-table reads) | **✅** | **NA** | ✗ | ◐ |
| **Spark-free** Iceberg ingestion (skade `IcebergSource`, no JVM) | **✅** | **NA** | ✗ | ✗ |
| Bulk columnar `RecordBatch` → **live graph DB** (FalkorDB UNWIND/CREATE) | **✅** | ◐ | ✅ | ✅ |
| Mature graph-algorithm library + production graph engine | ◐ | ✗ | **✅** | **✅** |

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

> **New here? → [knut for dummies](.nornir/knut-for-dummies.md)** — 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/`](.nornir/): the [map](.nornir/design.md) and
> one *territory* doc per subsystem —
> [graphar](.nornir/graphar.md) ·
> [graphar-falkordb](.nornir/graphar-falkordb.md) ·
> [graphar-flight](.nornir/graphar-flight.md) ·
> [knut-bifrost](.nornir/bifrost.md) ·
> [knut-pipelines](.nornir/pipelines.md). Locked decisions and open work are in
> [`.nornir/plan.md`](.nornir/plan.md); rendered docs (PDF + HTML) in [`docs/`](../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 `RecordBatch`es 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

| Crate | Description |
|---|---|
| [`graphar`](#graphar) | GraphAr gar/v1 format — read & write chunked property graphs (5 formats, local or object store) |
| [`graphar-falkordb`](#graphar-falkordb) | Bulk-load Arrow `RecordBatch`es into FalkorDB |
| [`graphar-flight`](#graphar-flight) | Arrow Flight SQL server — Cypher in, Arrow out; auth (bearer / Basic-handshake / mTLS); graph → lakehouse export (GraphAr file or Iceberg table) |
| [`knut-bifrost`](bifrost-design.md) | Bifröst 🌈 — watch Iceberg (Spark Connect *or* Spark-free via skade) or file drops, transform (incl. changelog CDC), sync to FalkorDB; Airflow AIP-82; file-or-Valkey state |
| [`knut-pipelines`](crates/knut-pipelines/docs/pipelines-viewer.md) | Viewer for Spark Declarative Pipelines + Spark Connect jobs — live DAG (egui-snarl) + job progress; `knutpipe` CLI/GUI |
| `xtask` | nornir bench suite (`cargo bench -p xtask`) |

### Python client

| Package | Description |
|---|---|
| [`knut`](#python-client-knut) | Flight SQL client — Cypher queries → pandas → NetworkX |

---

## Prerequisites

| Dependency | Version | Notes |
|---|---|---|
| Rust | stable ≥ 1.85 | Rust 2024 edition |
| Python | ≥ 3.11 | for the `knut` client |
| FalkorDB | latest | Redis/Valkey graph module |

**Start FalkorDB with Docker:**

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

---

## Quick start

### 1. Write a graph to disk (GraphAr format)

```rust
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

```rust
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

```rust
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

```python
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](https://graphar.apache.org/docs/specification/format) 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

```rust
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

```rust
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

| Format | Read | Write |
|---|---|---|
| Parquet | ✓ | ✓ |
| CSV | ✓ | ✓ |
| ORC | ✓ | ✓ |
| JSON (lines) | ✓ | ✓ |
| Arrow IPC | ✓ | ✓ |

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 `RecordBatch`es into a [FalkorDB](https://www.falkordb.com/) 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:

```cypher
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:

```cypher
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

```rust
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

| Arrow type | Cypher literal |
|---|---|
| `Int32` / `Int64` | integer, e.g. `42` |
| `Float32` / `Float64` | float, e.g. `3.14` |
| `Boolean` | `true` / `false` |
| `Utf8` | quoted string, e.g. `'Alice'` |
| `null` | `null` |

---

## `graphar-flight`

An [Apache Arrow Flight SQL](https://arrow.apache.org/docs/format/FlightSql.html) gRPC server that accepts **Cypher** queries instead of SQL, executes them against FalkorDB via `GRAPH.QUERY`, and streams results as Arrow `RecordBatch`es.

Any Flight SQL-compatible client works: [ADBC](https://arrow.apache.org/docs/format/ADBC.html), 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:

| Mode | How | When to use |
|---|---|---|
| **Auto** (default) | Column names and types inferred from the FalkorDB response. `Int` → `Int64`, `Double` → `Float64`, `Boolean` → `Boolean`, strings → `Utf8`. | Works for any query with zero setup. |
| **Registered** | Schema pre-declared via `register_vertex_query`, `register_edge_query`, or `register`. | When you want guaranteed types regardless of data, or need to catch schema drift early. |

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`.

```rust
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:

| Directive | Action |
|---|---|
| `VERTICES:<label>` | Insert vertices from the Arrow stream |
| `EDGES:<src>:<type>:<dst>` | Insert edges (stream must have `src` + `dst` Int64 columns) |

### 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.

| Redis RESP value | Arrow column type |
|---|---|
| `Int` | `Int64` |
| `Double` | `Float64` |
| `Boolean` | `Boolean` |
| `BulkString` / `SimpleString` | `Utf8` |
| `Nil` | `null` (type inferred from next non-null row; falls back to `Utf8` if all null) |

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](https://networkx.org/) 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

```bash
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.

```python
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

```python
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

```python
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

| Method | Description |
|---|---|
| `KnutClient(host, port, tls)` | Connect to a `graphar-flight` server |
| `query(cypher)` | Run a registered Cypher query → `pyarrow.Table` |
| `query_df(cypher)` | Run a registered Cypher query → `pandas.DataFrame` |
| `insert_vertices(label, df)` | Bulk-insert vertices via DoPut |
| `insert_edges(src, type, dst, df)` | Bulk-insert edges via DoPut |
| `to_networkx(vertex_query, edge_query, ...)` | Fetch graph → `networkx.Graph` |

### Tests

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

---

## Building & testing

```bash
# 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

```bash
# 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`](iceberg-falkordb-guide.md). Full design:
[`bifrost-design.md`](bifrost-design.md).

```yaml
# 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 }
```

```rust
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/`](containers/).

| Layer | Unit tests | Live containers (`--ignored`) |
|---|---|---|
| `graphar` formats | round-trip × {parquet, csv, json, orc, arrow-ipc} + empty-input | — |
| `graphar` metadata | YAML round-trips, chunk-path formatting (vertex/edge/offset/property) | — |
| `graphar` merge | two-group row-aligned merge, missing-column errors | — |
| `graphar` storage | `ChunkStore` local round-trip × 5 formats, raw bytes | **SeaweedFS** + **RustFS** S3 round-trips × 5 formats |
| `graphar-falkordb` | Cypher rendering, UNWIND query shape, batch splitting | live load + Cypher count/property checks vs **FalkorDB** and **FalkorDB-on-Valkey 9** (`KNUT_FALKORDB_IMAGE`) |
| `graphar-flight` | auto-schema inference, conversion, registry, named-table resolution, `SELECT … FROM` rewrite, auth (bearer/Basic/open), `skade` Iceberg-export round-trip (no container) | **Power BI path**: live FalkorDB + Basic handshake → bearer, `GetCatalogs`/`GetDbSchemas`/`GetTables`/`GetSqlInfo`, `SELECT * FROM <named query>` via the stock `FlightSqlServiceClient`; graph → GraphAr export round-trip |
| `knut-bifrost` | transforms, mapping YAML→pipeline, custom registry, state ledger, `FileSource` scan/ledger, Airflow asset-event model, **CDC change-row partitioning + `DETACH DELETE`**, **`skade` `IcebergSource`** read + snapshot tracking (no container), **create→detect→ingest lifecycle matrix** + a subprocess test that drives the `bifrost` CLI and asserts its JSON log events | **TOP GUN**: Spark 4.1.2 Connect + Iceberg 1.11.0 → bifröst → FalkorDB (initial + idle + incremental); **OSM** `relations.parquet` → `FileSource` → FalkorDB; real-FalkorDB lifecycle; **Valkey shared-ledger** across instances |
| `knut-pipelines` | DAG model (validate/topo), progress + `ViewerState` folding, proto conversion, snarl build | **live Spark 4.2**: `execute_sql` job progress + a full `define_graph → StartRun` SDP run (QUEUED→…→COMPLETED) |

```bash
# 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
```

[testcontainers]: https://crates.io/crates/testcontainers

## Benchmarks

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

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

---

## Roadmap

| Item | Status |
|---|---|
| GraphAr Parquet + CSV read/write | ✓ done |
| FalkorDB bulk insert via Cypher UNWIND | ✓ done |
| Arrow Flight SQL server (Cypher dialect) | ✓ done |
| Python client → NetworkX | ✓ done |
| `merge_property_groups` / `VertexReader::read_merged` | ✓ done |
| ORC + JSON + Arrow IPC file formats | ✓ done |
| Object storage (S3 / GCS / Azure / SeaweedFS / RustFS via `object_store`) | ✓ done |
| Flight SQL authentication (TLS + bearer token) | ✓ done |
| Bifröst: Spark Connect + Iceberg → transform fabric → FalkorDB | ✓ done |
| Container test matrix (FalkorDB-on-Valkey 9, Spark 4.1.2 + Iceberg 1.11, SeaweedFS, RustFS) | ✓ done |
| nornir bench suite | ✓ done |
| Airflow 3 asset events (AIP-82): emit + `FileSource` listener | ✓ done |
| Power BI / ADBC via Flight SQL metadata surface (catalogs, schemas, tables, SqlInfo, prepared statements) | ✓ done |
| Flight SQL auth: bearer + Basic/handshake (Power BI/ODBC) + mutual TLS | ✓ done |
| Bifröst changelog CDC (Iceberg `_change_type` → upserts + `DETACH DELETE`) | ✓ done |
| Spark-free Iceberg source via `skade` (`IcebergSource`, snapshot-incremental) | ✓ done |
| Graph → lakehouse export: GraphAr file + Iceberg table (via `skade`) | ✓ done |
| Valkey-backed shared `SyncState` + graceful shutdown | ✓ done |
| `Date`/`Timestamp`/`List` → Cypher rendering | ✓ done |
| Spark Declarative Pipelines viewer (`knutpipe pipeline`: define + run + stream) | ✓ done |
| WASM / Flight transform plugins; REST Iceberg catalog; Iceberg-table CDC export | proposed |

The full proposition list with rationale: [`.nornir/plan.md`](.nornir/plan.md).

---

## License

Apache-2.0