nordisk/knut

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; the why in 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.

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:

// 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):

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.

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' ids, 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:

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

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

f2b — feeding it Arrow you already have (ArrowSource)

You do not need a Spark endpoint or a warehouse directory. If something already hands you RecordBatches — a warehouse client, a Flight read, an ADBC result, a query you ran yourself — that is a source:

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:

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:

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

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.