# Bifröst — tables → graph, guide
> **Class:** `guide` · audience: a techie writing a MappingSpec or driving `bifrost`.
> Reuse: `bifrost.uses = [falkordb, graphar]`.
> The *what* is in [`bifrost-design.md`](bifrost-design.md); the *why* in [`bifrost-reasoning.md`](bifrost-reasoning.md). This file is the workflow.
## f0 — zero to a running listener (the whole workflow)
Four moves take a table to a live graph:
1. **Write a `MappingSpec` YAML** — one entry per table: its `mode` (`full` /
`incremental` / `cdc`) and a `pipeline` of `vertex` / `edge` / `fk_edges` / `custom`
steps.
2. **Pick a source** — `FileSource::new(dir)` (parquet drops), `IcebergSource::new(warehouse)`
(feature `skade`, Spark-free), or `SparkConnectSource::connect(url)` (feature `spark`).
3. **Pick a sink + ledger** — `FalkorDbSink::connect(url, graph)` and `SyncState::open(path)`
(or `open_valkey` to share a cursor across instances).
4. **Build and drive a `Listener`** — `Listener::new(spec, registry, source, sink, state)`,
then `sync_once()` once or `run(interval, on_round)` forever.
The CLI (`bifrost`, feature `cli`) wires all four for a `FileSource` so you can try it with
no code.
> ⚠ **…and `FileSource` is all it wires.** `src/bin/bifrost.rs:230` constructs
> `FileSource::new(&files)` and nothing else — there is no `--warehouse`, `--iceberg` or
> `--spark` flag. **Both Iceberg sources are library-only**: to stream an Iceberg table into
> a graph you write a Rust `main` (§f2, `watch_iceberg_delta`). Both Iceberg read lanes, the
> transfer-method menu into FalkorDB, and the no-pushdown caveat:
> [`iceberg-falkordb-guide.md`](iceberg-falkordb-guide.md).
Your choice in move 2 is also what decides how much infrastructure you have to start.
`IcebergSource` needs no service at all — it opens a local `skade` warehouse directory
(`catalog.redb` + `warehouse/`) in-process, no JVM, no catalog server. `SparkConnectSource`
needs a live Spark Connect endpoint with the Iceberg runtime jar, and is the only route to a
remote REST/Glue/Hive catalog or an S3-backed warehouse. Both end at a FalkorDB that must be
running. The full picture, including which boxes are processes and which are just linked
crates:
⚠ One operational constraint the map flags and the code repeats three times: the `skade`
warehouse is **redb-backed and single-open**, and `IcebergSource` reopens it on every poll. A
concurrent external writer holding the warehouse blocks the listener, so the Spark-free lane is
a hand-off lane (writer commits and closes, poller opens) — not a live shared-warehouse lane.
## f0b — the knut ▸ Spark 4.1 + Iceberg 1.11.0 LOAD demo
`src/spark_demo.rs` is the demo half of the Spark path: the **one source of truth** for
"land a real Iceberg table on the blessed Spark 4.1.2 GA + `iceberg-spark-runtime-4.1_2.13:1.11.0`
pair", shared by knut's own tests, korp's 🧰 Tools menu, and the container integration
tests — so nobody keeps a second copy of the SQL.
Two halves, deliberately split so the wiring is provable without a container:
```rust
// PURE + feature-free — build and assert this anywhere, no Spark, no network.
let plan = knut_bifrost::spark_demo::iceberg_demo_plan("sc://localhost:15012");
assert_eq!(plan.table, "lake.knut_demo.people");
// feature `spark` — EXECUTE it over the unchanged SparkConnectSource.
let rows = knut_bifrost::spark_demo::run_iceberg_demo(&plan).await?; // -> 8
```
`iceberg_demo_plan` builds the ordered program `DROP → CREATE NAMESPACE → CREATE TABLE
… USING iceberg → INSERT` plus two verifications: a `SELECT count(*)` (must equal
`plan.rows` exactly — the row set is deterministic, no rng, no timestamps) and a
`SELECT snapshot_id FROM <table>.snapshots`, which is the exact metadata path
`SparkConnectSource::fetch_new` polls. A green demo therefore also proves the streaming
listener's poll path is live, not just that Spark answered.
`run_iceberg_demo` fails **loud**: a missing Iceberg runtime surfaces as the CREATE
TABLE error, a wrong count as an explicit mismatch, a missing snapshot as "is the
Iceberg runtime loaded?" — never a silent success.
It never starts a container. In the constellation the Spark 4.1 + Iceberg server is
brought up THROUGH jera (korp's `infra::spark_iceberg_spec()` → `start_via_jera` →
draupnir's one OCI engine), and korp surfaces the whole thing as the selectable
🧰 Tools entry **"🧊 knut ▸ Spark 4.1 + Iceberg 1.11.0 — load a lake table"**. knut is
headless, so it has no Tools menu of its own — see korp's `.nornir/tools-menu-design.md`.
Knobs: `iceberg_demo_plan_for(url, catalog, namespace, table, rows)` scopes a second
zone onto the same program instead of a copied SQL string.
## f1 — the MappingSpec (YAML)
Loaded by `MappingSpec::from_yaml(str)` / `load(path)` (`src/mapping.rs:92`/`:96`). A worked
example (`src/mapping.rs` documents the grammar):
```yaml
version: 1
tables:
- table: lake.social.users
mode: incremental # snapshot-diff delta (Iceberg) / new files (FileSource)
pipeline:
- vertex:
label: Person
id_column: user_id
properties: [name, age, city]
rename: { city: location }
- fk_edges: # FK pair on the same row → an edge
type: WORKS_AT
src: { label: Person, column: user_id }
dst: { label: Company, column: company_id }
- table: lake.social.follows
mode: full # small table, re-merge each change
pipeline:
- edge:
type: FOLLOWS
src: { label: Person, column: follower_id }
dst: { label: Person, column: followee_id }
properties: [since] # edge props → SET e += r.props
- table: lake.crm.orders
mode: cdc # Iceberg changelog: inserts, updates, deletes
pipeline:
- custom: order_normaliser # a Rust Transform registered by name
```
`vertex` **and** `edge`/`fk_edges` take `properties: [...]` + optional `rename`; edge
property columns become relationship properties. An edge's optional `key: [col, …]`
joins the relationship's MERGE key, so rows with the same endpoints and different keys are
**parallel** edges (without it: one edge per `(src, type, dst)`).
**Deletes** (`mode: cdc`): a `DELETE` row removes the node its `vertex` step names
(`DETACH DELETE`, taking its edges) **and** the edge its `edge` step names (`DELETE e`) — so
an edge-only table (`follows`) deletes its edges, and a row whose foreign key changed drops
the edge to the old endpoint. A node or edge also upserted in the same round survives (an
UPDATE is a delete plus an insert).
**A changelog that carries only the key** — an equality delete, then the new row, as
key-based upsert writers do — has no old foreign key and no old endpoints. Two rules cover it:
* an `fk_edges` step on an **entity row** (the table's `vertex` step is the edge's source:
same label, same id column) is ONE edge per source node: before the upsert, the source's
other outgoing edges of the type are deleted, and all of them when the foreign key is
cleared (null). The new row is the truth, whatever the old one carried. A junction table
(no such vertex, many rows per source) keeps every edge. `replace: true|false` on any edge
step overrides;
* an edge whose `DELETE` row has null endpoints but a `key` is deleted by that key
(`MATCH ()-[e:T {k: …}]->() DELETE e`, the key indexed), before the round's upserts.
Spark (4.1) writes an UPDATE/DELETE/MERGE as the full old row — copy-on-write rewrites the
file, merge-on-read on format v2 writes position deletes — so its changelog always names
the old edge. The exception is a **format-v3** table with merge-on-read: Iceberg writes
**deletion vectors**, which skade does not read as rows yet, so such a commit re-reads the
table in full. Keep MoR tables on v2 (or copy-on-write) for the delta path.
**Into FalkorDB: first load in bulk, then MERGE.** A label empty in the graph the first time a
round writes it (and whose round deletes nothing) is loaded with a verbatim `GRAPH.BULK` —
rows deduplicated on the id, last wins; no `_gar_id` property. Every later write is
`UNWIND … MERGE`, so replays, updates and deletes are always idempotent. A bulk the server
refuses falls back to MERGE (same graph). **Edges too:** the first write of a relationship type
with no edges yet (a round that deletes nothing) is a `GRAPH.BULK` append — the end nodes'
internal ids looked up once per label (`MATCH (n:L) RETURN n.id, ID(n)`), rows deduplicated on
`(src, dst, key)`; a row whose end node does not exist yet stays on MERGE, which creates it.
Every MERGE key — `(label, id)` of nodes and edge endpoints — is indexed once per label.
Measured (`tests/first_load_bulk_falkordb.rs`, release, 200 000 each): nodes `GRAPH.BULK`
318 ms vs indexed MERGE 10 031 ms (31×); edges 975 ms vs 27 210 ms (28×).
## f1b — three mapping formats, one MappingSpec
`--mapping` takes any of three formats, told apart by extension; each compiles
(`trait GraphMapping`, `src/mapping.rs`) to the MappingSpec above, so modes, cursors,
deletes and sinks are the same whichever it is. The two that name no sync mode take
`--mode cdc|incremental|full` (default `cdc`) for every table.
| file | format | impl |
|---|---|---|
| `.yml` / `.yaml` / `.ron` | bifröst's own, above: per table, a pipeline of steps | `MappingSpec` |
| `.json` | a **PuppyGraph** `schema.json` (graph-first `node` + `edge` lists) | `PuppyGraphMapping` (`src/puppygraph.rs`) |
| `.sql` | one **SQL** `SELECT` per step (feature `sql`) | `SqlMapping` (`src/sql.rs`) |
**PuppyGraph.** Each `node` becomes a `vertex` step and each `edge` an `edge` step of the
table its `externalDataSource` names (`schema.table`); `mappedField` is the rename, a node's
`id` its MERGE key, an edge's `fromKey`/`toKey` are matched on the end nodes' `id`s, and an
edge's `id` is its `key` (parallel edges). The `catalog` section is not read — the lake is
`--warehouse` / `--s3-root` / `--rest-uri` / `--grpc-uri`. Not yet: composite keys, and the older
`oneToOne`/`tableSource` layout (both refused by name).
**SQL.** A statement reads ONE table (its `FROM`; no joins, `UNION` or `WITH`) and runs over
each round's rows in DataFusion; the reserved columns it returns say what a row becomes:
```sql
-- nodes: _label + _id (keyed on property `id`); every other column is a property
SELECT CASE WHEN staff THEN 'Staff' ELSE 'Person' END AS _label, user_id AS _id,
upper(name) AS name FROM social.users WHERE active;
-- edges: _type, _from_label, _from, _to_label, _to (endpoints matched on `id`);
-- optional _edge_id is the edge key (parallel edges)
SELECT 'WORKS_AT' AS _type, 'Person' AS _from_label, user_id AS _from,
'Company' AS _to_label, company_id AS _to, since FROM social.users;
```
A step can also sit in a YAML pipeline: `- sql: "SELECT … FROM src"`. **Deletes run the
statement without its `WHERE`**: a key-only `DELETE` row has every other column null, and
the filter would drop it — so compute `_label`/`_id` (and edge endpoints) from key columns
or constants.
## f2 — the Rust API
```rust
use knut_bifrost::{Listener, MappingSpec, TransformRegistry, SyncState, FileSource, FalkorDbSink};
let spec = MappingSpec::load("mapping.yml")?;
let mut reg = TransformRegistry::new();
reg.register("order_normaliser", Arc::new(MyTransform)); // custom steps, before start
let source = FileSource::new("./drops");
let sink = FalkorDbSink::connect("redis://127.0.0.1:6379", "demo").await?;
let state = SyncState::open("./.bifrost-state.json")?; // or SyncState::open_valkey(url, key).await?
let mut listener = Listener::new(spec, reg, source, sink, state)
.with_asset_emitter(Arc::new(AirflowEmitter::new("http://airflow", token)), "iceberg://lake/");
listener.sync_once().await?; // one pass
listener.run(Duration::from_secs(30), |r| !r.is_failure()).await?; // loop w/ implicit backoff cap
listener.run_with_backoff(Backoff::new(iv, max), on_round).await?; // explicit cap
```
| you need… | call |
|---|---|
| custom transform | `TransformRegistry::register(name, Arc<dyn Transform>)` (`src/registry.rs`) |
| WASM transform | `WasmTransform::from_file(path, label, id_column)` → register as `custom` (`src/wasm.rs:67`) |
| Flight transform | `FlightTransform::connect(endpoint, label, id_column).await` (`src/flight.rs:65`) |
| shared cursor | `SyncState::open_valkey(url, key).await` (`src/state.rs:79`) |
| Spark-free delta stream | `watch_iceberg_delta(spec, warehouse, url, graph, state, interval, on_round)` (`src/listener.rs:372`) |
| live counters | `listener.metrics()` (`src/listener.rs:109`) + `render_prometheus(m)` (`src/metrics.rs:75`) |
## f2b — feeding it Arrow you already have (`ArrowSource`)
You do not need a Spark endpoint or a warehouse directory. If something already hands
you `RecordBatch`es — a warehouse client, a Flight read, an ADBC result, a query you
ran yourself — that is a source:
```rust
use knut_bifrost::{ArrowTables, Listener, MappingSpec, SyncState, TransformRegistry};
use knut_bifrost::source::ArrowSource;
// batches from wherever: skade over REST, an Iceberg REST catalog, Spark in-process
let tables = ArrowTables::new().with_table("lake.hr.people", batches);
let source = ArrowSource::new("skade-remote", tables);
let mut listener = Listener::new(spec, TransformRegistry::new(), source, sink, state);
listener.sync_once().await?;
```
`ArrowSource::from_batches(producer, table, batches)` is the one-table shorthand.
**Three things to know.**
1. **A zero-row read is refused, by default.** Not `Ok(None)` — an error naming the
table, the producer, and the tables the producer actually holds. If the table is
genuinely empty, say so: `ArrowSource::new(..).allowing_empty()`.
2. **`SyncMode::Cdc` is refused.** An Arrow batch has no snapshot lineage to diff.
Use `SyncMode::Full` for a complete rebuild; for real CDC use `SparkIcebergCdc`,
`SparkNativeCdc`, or `IcebergSource` over a local skade warehouse.
3. **`SyncMode::Incremental` ships once** and then reports nothing, for the same
reason. Full mode re-reads every round; the sink's MERGE makes that safe.
For a one-shot graph with no listener, sink or ledger, use the offline seam instead:
```rust
let (mutation, report) = knut_bifrost::build_mutation_reported(&spec, &tables)?;
report.require_non_empty()?; // an empty build is refusable in one line
```
## f3 — the `bifrost` CLI (feature `cli`)
`src/bin/bifrost.rs`. The subcommands:
```bash
# stage a parquet drop so `run` needs no other tooling
bifrost gen-sample --files ./drops --table users --rows 10000 [--chunk chunk0.parquet]
# one pass, dry (no DB): the counting null sink, JSON logs
bifrost run --mapping m.yml --files ./drops --dry-run --once --json-logs
# poll forever until Ctrl-C, live metrics, emit Airflow events
bifrost run --mapping m.yml --files ./drops --graph demo --falkor redis://127.0.0.1:6379 \
--interval 30 [--max-backoff 300] [--state-valkey URL --state-key K] \
[--metrics-file out.prom] [--metrics-addr 127.0.0.1:9090] [--emit-airflow http://airflow …]
bifrost self-test # headless transform→mutation invariant check, no DB
# stream Iceberg into the graph (feature skade; S3 needs skade-s3, REST skade-rest, gRPC skade-grpc)
bifrost watch --warehouse /var/lib/lake --mapping m.yml --graph g # skade warehouse dir
bifrost watch --s3-root s3://lake/wh --s3-endpoint https://minio:9000 \
--mapping m.yml --state st.json # S3, no catalog
bifrost watch --rest-uri https://minio:9000/_iceberg --rest-warehouse analytics \
--sigv4 s3tables --s3-endpoint https://minio:9000 --mapping schema.json # MinIO AIStor
bifrost watch --grpc-uri http://katalog:8182 --grpc-namespace social \
--mapping m.yml --state st.json # skade-katalog gRPC
```
`--s3-root`: a mapping table `ns.name` is the table root `<root>/ns/name` (an `s3://…`
mapping table is used as-is), its metadata `version-hint.text`, else the newest
`*.metadata.json`. `--rest-uri`: an Iceberg REST catalog; AIStor's signs with SigV4 service
`s3tables`. `--grpc-uri`: a skade-katalog gRPC front (`skade-katalog serve --grpc ADDR`, plain
HTTP/2), the same catalog its REST front serves, one RPC per table load; the tables' files
come from S3 when `--s3-endpoint` is given, else from this box. S3 keys come from
`MINIO_ACCESS_KEY` / `MINIO_SECRET_KEY` only, never a flag; a private CA is `--s3-ca PEM` or
`BIFROST_S3_CA_PEM`.
`--interval` polls until **Ctrl-C** (graceful: finishes the round + persists, then exits);
failing rounds back off up to `--max-backoff` (default 10× interval). `--metrics-addr` needs
build feature `metrics-http`; absent → no server.
## The log contract (drive it without a human)
Every stage emits a structured tracing event, so a test/bencher can drive `bifrost` (with
`--json-logs`) and assert the lifecycle from stdout — no DB, no eyes:
| `fields.event` | target | when |
|---|---|---|
| `file_detected` | `bifrost::file` | a new chunk file is seen (per file) |
| `files_read` | `bifrost::file` | the round's new files are read |
| `dry_run_mutation` | `bifrost::sink` | `--dry-run`: mutation counted, not written |
| `sink_applied` | `bifrost::sink` | a `GraphMutation` was MERGE'd into FalkorDB |
| `vertices_deleted` | `bifrost::sink` | CDC `DETACH DELETE` applied (`count`) |
| `table_synced` | `bifrost::round` | one table ingested (`rows`, `vertices`, `edges`) |
| `round_complete` | `bifrost::round` | sync round done (`synced`, `failed`) |
| `backoff` | `bifrost::round` | a no-progress round backed the next sleep off (`sleep_secs`, `interval_secs`) |
| `assets_emitted` | `bifrost::airflow` | AIP-82 events emitted (`count`) — when an emitter is wired |
| `metrics` | `metrics` | cumulative counters after every round (`rounds_total`, `rounds_failed`, `tables_synced`, `rows_ingested`, `vertices_written`, `edges_written`, `deletes`, `assets_emitted`) |
| `metrics_written` | `metrics` | `--metrics-file`: Prometheus text rendered on shutdown (`path`) |
| `metrics_http_started` | `metrics` | `--metrics-addr` (feature `metrics-http`): live `/metrics` server bound (`addr`) |
The `metrics`/`backoff` events are **purely additive** — they ride *after*
`round_complete`/`assets_emitted` and never alter them, so the log order above is stable.