# Bifröst — tables → graph, design
> **Class:** `design` · **Function axis:** `f0` (the fabric) + `f1` sources · `f2` transform fabric · `f3` MERGE sink · `f4` state ledger · `f5` CDC · `f6` Airflow · `f7` metrics/backoff.
> Reuse: `bifrost.uses = [falkordb, graphar]` — it materialises into a FalkorDB graph and speaks the same `_gar_id`/GraphAr id convention.
> Strict: this file states *only the shape we run*. Every *why*, alternative, and honest limitation lives in [`bifrost-reasoning.md`](bifrost-reasoning.md); the workflow in [`bifrost-guide.md`](bifrost-guide.md).
## f0 — the fabric (the whole)
Bifröst is the bridge from the table world to the graph world: a **`TableSource`** polls
rows added since a per-table cursor, a **transform fabric** turns those rows into a
**`GraphMutation`**, and an idempotent **`GraphSink`** MERGEs it into FalkorDB. Three traits,
one currency:
| trait | signature | where |
|---|---|---|
| **`TableSource`** | `async fn fetch_new(&self, &TableSpec, &TableState) -> Result<Option<(Vec<RecordBatch>, TableState)>>` | `src/source.rs:TableSource` (`:18`) |
| **`Transform`** | `fn apply(&self, &TableContext, &[RecordBatch]) -> Result<GraphMutation>` | `src/transform.rs:Transform` (`:26`) |
| **`GraphSink`** | `async fn apply(&mut self, &GraphMutation) -> Result<SinkReport>` | `src/sink/graph.rs:GraphSink` (`:16`) |
The currency is **`GraphMutation`** (`src/mutation.rs:GraphMutation`, `:55`): `vertices:
Vec<VertexSet>` (label, id_column, batch), `edges: Vec<EdgeSet>` (src/type/dst + batch),
`deletes: Vec<VertexDelete>`. Arrow `RecordBatch`es flow source→transform→sink with no
conversion. One round = `src/listener.rs:Listener::sync_once` (`:127`): for each mapped
table, `fetch_new` → run the compiled pipeline → `sink.apply` → **advance state**.
The laws that make the loop hold (a round is conformant iff all hold):
| law | statement | enforced by |
|---|---|---|
| **state-after-sink** | per-table cursor is written **only after** the sink succeeds; a crash replays the round, MERGE dedupes | `Listener::sync_once` advances `SyncState` last (`listener.rs:127`) |
| **MERGE both endpoints** | every edge MERGEs *both* nodes, so an edge landing before its owning table leaves an enrichable placeholder | `src/sink/graph.rs:FalkorDbSink::upsert_edges` (`:56`) |
| **vertices → edges → deletes** | one round applies node upserts, then edge upserts, then `DETACH DELETE`s — same-round upsert+delete resolves to *gone* | `FalkorDbSink::apply` order (`sink.rs`) |
| **idle ≠ failure** | a round with nothing synced and nothing failed keeps base cadence; only ≥1 attempt with *all* failing is a failure | `src/listener.rs:RoundReport::is_failure` (`:35`) |
## f1 — sources (`TableSource`)
`src/source.rs`. `TableSpec { table, mode }` (`:11`) names a table and its `SyncMode`.
Four implementors, all behind the one trait:
| source | ctor | feature | reads |
|---|---|---|---|
| **`SparkConnectSource`** | `connect("sc://host:15002")` (`source.rs:205`) | `spark` | Iceberg via Spark Connect gRPC; snapshot polling, `EXCEPT ALL` incremental, `system.create_changelog_view` CDC (`:288`) |
The `spark` feature is served by the **in-tree vendored `spark-connect` fork**
(`vendor/spark-connect-arrow59/`, patched in via root `Cargo.toml`
`[patch.crates-io]`) — upstream `spark-connect` 0.2.2 with arrow bumped 56 → 58
and **no source changes**, so bifröst consumes its batches as workspace Arrow 58
directly (no IPC round-trip bridge). The fork is fully self-contained: all seven
`protobuf/spark/connect/*.proto` are committed and `build.rs` compiles them with a
*vendored* protoc (`protoc_bin_vendored`), so `git clone knut && cargo build -p
knut-bifrost --features spark` needs **no external checkout, no system protoc, no
network**.
| **`IcebergSource`** | `new(warehouse_dir)` (`source.rs:509`) | `skade` | Spark-free Iceberg via skade's embedded catalog; snapshot-id polling, single Arrow-58 tree (plain `clone`, no FFI) (`:749`) |
| **`FileSource`** | `new(dir)` (`source.rs:59`) | — | parquet drops under `<dir>/<table>/`; ledger `done_files` gates Incremental vs Full (`:105`) |
| **`ArrowSource`** | `new(producer, batches)` / `from_batches(producer, table, batches)` | — | **Arrow batches a caller already holds.** Any producer implementing `ArrowBatches`; `rel2graph::ArrowTables` is one, so the same in-memory holder drives both the offline `build_mutation` and the live `Listener` |
### `ArrowSource` — the source for warehouses bifröst cannot open itself
`SparkConnectSource` needs a Spark Connect **endpoint**. `IcebergSource` needs a local
skade warehouse **directory**. Neither can be pointed at a warehouse reached any other
way: a remote skade over its Iceberg-REST front, an Iceberg REST catalog, a client
living in the calling process. All of those already produce **Arrow**, and that is
what `ArrowSource` takes.
It is built ON `rel2graph::ArrowTables`, not beside it. `ArrowTables` was already the
crate's answer to "some Arrow I already have"; it fed only the *offline*
`build_mutation`. Implementing `ArrowBatches` for it is what lets the same bag of
batches drive the round loop — the sink, the state ledger, the asset events.
**It is a batch read, not CDC**, and the code says so rather than only the docs:
| `SyncMode` | `ArrowSource` |
|---|---|
| `Full` | re-reads the producer every round (idempotent MERGE makes that safe) |
| `Incremental` | ships once, then reports nothing — there is no delta to compute |
| `Cdc` | **typed refusal**, naming `SparkIcebergCdc` / `SparkNativeCdc` / `IcebergSource` as what can |
`cdc_capability()` is `None`, the same honest answer `FileSource` gives, so an
operator reading `state_json` is not misled either. **skade reached this way and
iceberg-rest feed bifröst for a full build, not for streaming.**
**An empty read is not a built graph.** `EmptyRead::Refuse` is the default: zero
batches, or batches carrying zero rows, is a typed `BifrostError::Source` naming the
table, the producer, what it actually got, and — when the producer can enumerate —
the tables it *does* hold. `allowing_empty()` is the explicit opt-out. The hazard is
real and was measured: with the guard removed, an empty skade warehouse produced a
round with **zero failures and zero vertices**, indistinguishable from a successful
build.
The same hazard exists at the offline seam, where `build_mutation` silently `continue`s
past a spec table the source has no rows for. `build_mutation_reported` returns a
`BuildReport { tables_mapped, tables_empty, rows_read, vertices, edges }` beside the
mutation, and `BuildReport::require_non_empty()` is the one-line refusal.
`build_mutation` delegates to it, so the counts can never describe a different run
than the mutation they came with.
**Spark-free delta** — `IcebergSource` in Incremental mode reads **only the data files
appended since the cursor**, a manifest-level snapshot diff (skade's `Table::read_delta`).
`ingest_delta(table, from, to) -> (Vec<RecordBatch>, DeltaPlan)` (`source.rs:546`) is the
one-shot window; `DeltaPlan { snapshots, added_files, delete_files, needs_full_reload }`
(`source.rs:491`) reports what the delta resolved to. `latest_snapshot(table)` (`:672`) is
the change detector. The stream loop is `src/listener.rs:watch_iceberg_delta` (`:372`),
which wires `IcebergSource` + `FalkorDbSink` + a file cursor into a `Listener` and polls via
`sync_once` — same fabric, same MERGE, same asset events. Full-reload is triggered only by
overwrite/replace/position-delete snapshots.
## f2 — the transform fabric
`src/transform.rs` + `src/mapping.rs`. A pipeline is a list of named `PipelineStep`s
(`mapping.rs:PipelineStep`, `:74`) compiled against a `TransformRegistry` by
`MappingSpec::compile` (`mapping.rs:103`). All steps in one pipeline see the **same** input
batches; their mutations concatenate (`GraphMutation::merge`, `mutation.rs:71`).
**RON parity + schema→mapping codegen (the closed Iceberg→graph seam).** The
`MappingSpec` was YAML-only; it now round-trips **RON** too — `from_ron` / `to_ron`
/ `load_ron` (`mapping.rs`) beside `from_yaml`, over the same derives (YAML stays
accepted; additive). This makes the Iceberg→graph mapping a first-class RON object
like every other spec in the constellation, and RON is what the codegen emits.
`MappingSpec::from_arrow_schema(table, &Schema)` derives a **candidate**
`MappingSpec` from an Iceberg table's Arrow schema — one vertex keyed on a guessed
id, scalar columns as properties, one `fk_edges` candidate per `*_id` column —
returned as a `SchemaMappingCandidate { spec, unmapped, guesses }`. It is **honest
by construction (the findings-doc law): no column is ever silently dropped** —
every field is mapped (id / property / FK endpoint) or SURFACED in `unmapped` (a
nested/complex Arrow type that cannot be a scalar graph property), and every
guessed role is listed in `guesses` for a human to review. Emit it with
`to_ron()`, correct by hand, load it back. Covered by `mapping::ron_seam_tests`
(YAML↔RON parity; schema→candidate; nested-column-surfaced-not-dropped). *Open
design decision left explicit (not made implicitly here): whether this RON dialect
folds into `runa` or stays knut-local, and whether to teach thund an Iceberg source
vs route thund→bifrost — see the workspace findings doc.*
| kind | fabricates | where |
|---|---|---|
| `vertex` | one row → one node; column subset + `rename` + `id_column` | `src/transform.rs:VertexTransform` (`:51`) |
| `edge` | one row → one relationship; property columns ride next to `src`/`dst` (`SET e += r.props`) | `src/transform.rs:EdgeTransform` (`:86`) |
| `fk_edges` | FK column pair on the row → edge (endpoints owned by other steps/tables) | same `EdgeTransform`, `PipelineStep::FkEdges` (`mapping.rs:83`) |
| `custom` | a `Transform` impl looked up by name | `src/registry.rs:TransformRegistry::register`/`get` |
Two out-of-process transform kinds register as `custom` and keep the same `GraphMutation`
contract — each output Arrow batch → one `VertexSet{label, id_column}`:
- **WASM** — `src/wasm.rs:WasmTransform::from_file(path, label, id_column)` / `from_bytes`
(`:67`/`:50`), feature `wasm`. A sandboxed `wasmi` guest, Arrow IPC in/out.
- **Flight** — `src/flight.rs:FlightTransform::connect(endpoint, label, id_column)` /
`from_client` (`:65`/`:85`), feature `flight`. Proxies rows to an external Arrow-Flight
`DoExchange` service (any language); `apply` bridges sync→async internally.
## f3 — the sink (`GraphSink` → FalkorDB)
`src/sink/graph.rs:FalkorDbSink` (`:21`), `connect(url, graph)` (`:27`), `with_batch_size(n)`
(`:34`). `apply` unwinds each set into Cypher: `upsert_vertices` (`:39`) MERGEs nodes on the
id key; `upsert_edges` (`:56`) MERGEs *both* endpoints then the relationship; the delete pass
(`:83`) `DETACH DELETE`s every id named by a `VertexDelete` (which also clears incident
edges, so there is no separate edge-delete set). Returns `SinkReport { vertices_written,
edges_written, deleted }` (`mutation.rs:80`).
**Null FK = no edge.** `rows_as_cypher_maps` drops any edge row whose `src`/`dst`
endpoint is NULL (a nullable FK with no target — a self-ref root, an optional FK),
so the sink never `MERGE`s a `null`-keyed node or a dangling edge into it. This
matches the offline correctness oracle (which excludes `null_endpoint_rows`), so
the live graph and the oracle agree; `edges_written` therefore counts real
relationships only. The whole `build_mutation → FalkorDbSink::apply → read-back`
path — with the composite-key junction, self-ref, null-FK, parallel-edge and
typed-property patterns — is proven live in
`tests/live_falkordb_sink.rs` (`#[ignore]`, throwaway graph; see
[rel2graph-engine-seam.md](rel2graph-engine-seam.md)).
## f4 — the state ledger (`SyncState`)
`src/state.rs:SyncState` (`:43`), two backends behind one API. `TableState` (`:26`) is the
per-table cursor: `snapshot_id` (Iceberg) + `done_files` (`#[serde(default)]` file ledger).
| backend | open | semantics |
|---|---|---|
| **File** | `SyncState::open(path)` (`state.rs:62`) | atomic-file ledger, single process |
| **Valkey** | `SyncState::open_valkey(url, key)` (`state.rs:79`) | shared Valkey-9+ hash; whole ledger loads on open, write-through per advance, so multiple instances share one cursor |
`get` (`:103`) is served from the in-memory cache; `advance` (`:121`) is async
(write-through). `Listener::sync_once` calls `advance` only after the sink succeeds (f0).
## f5 — CDC (`SyncMode::Cdc`)
`src/mapping.rs:SyncMode` (`:45`): `Full` (`:48`), `Incremental` (`:51`), `Cdc` (`:59`). In
`Cdc` the Spark source reads a changelog — from **Spark 4.2's native `CHANGES` clause**
or **Iceberg's `system.create_changelog_view`**, chosen by probing the live server
(`cdc-design.md` §f2) — adding a `_change_type` column
(`SyncMode::CHANGE_TYPE_COLUMN`, `mapping.rs:62`). Routing lives in
`src/transform.rs`: `partition_change_rows` (`:211`) splits the batch by
`_change_type` parsed through `cdc::ChangeOp`, so **both engine dialects**
(Iceberg's `UPDATE_AFTER`, Spark's `update_postimage`) route identically —
INSERT/UPDATE_AFTER → upserts, DELETE → deletes, UPDATE_BEFORE dropped — and
`deletes_from_change_rows` (`:274`) turns delete rows into `VertexDelete`s. The
sink's vertices→edges→deletes order (f0) makes a same-round upsert+delete resolve to *gone*.
The Spark-free `IcebergSource` produces the same `_change_type` column from equality-delete
files (f1), so CDC works with no Spark.
## f6 — Airflow AIP-82
`src/airflow.rs`. `round_to_events(uri_prefix, &RoundReport) -> Vec<AssetEvent>` (`:38`)
maps a finished round to one `AssetEvent` (`:20`) per ingested table (uri = `{prefix}{table}`);
`emit_round(&emitter, uri_prefix, &round)` (`:65`) POSTs them through an `AssetEmitter`
(`:30`). `AirflowEmitter::new(base_url, token)` (`:96`, feature `airflow`) is the HTTP
implementor hitting `{url}/api/v2/assets/events`. Wired in via
`src/listener.rs:Listener::with_asset_emitter(Arc<dyn AssetEmitter>, uri_prefix)` (`:116`):
every successful round auto-emits, closing ingested → asset event → DAG.
## f7 — metrics + backoff, the loop
`src/metrics.rs`. `Metrics` (`:25`) is a cumulative counter set folded after every round by
`record_round` (`:51`); `render_prometheus(&Metrics) -> String` (`:75`) is pure (no HTTP dep).
`Backoff::new(interval, max)` (`:156`) drives the failure schedule: `next_sleep(failed)`
(`:180`) doubles the sleep on a failing round (`interval → 2× → … → max`, default
`interval*10`) and resets on progress. The loop drivers are `Listener::run(interval, on_round)`
(`listener.rs:293`, implicit `interval*10` cap) and `run_with_backoff(Backoff, on_round)`
(`:311`, explicit cap); `Listener::metrics()` (`:109`) exposes the live counters. Live scrape
is `src/metrics_http.rs:MetricsServer::start(addr)` (`:91`, feature `metrics-http`) serving
`GET /metrics` off a `tiny_http` thread; the loop calls `handle().publish(metrics)` (`:63`)
after each round.
## Iceberg read pushdown — projection + predicate (2026-07)
`IcebergSource`'s `SyncMode::Full` path called `skade::Table::read()` on every
snapshot change: **every column of every data file**, however narrow the mapping.
skade had carried the pruning surface all along — `read_columns` pushes the
projection into the row-group readers (dropped columns are never decompressed or
allocated), `read_filtered` pushes a predicate into the scan planner (non-matching
data files are never opened), `plan_stats` costs a plan without reading it — and
bifrost called **none** of it.
### The ask travels on `TableSpec`
`TableSpec` gained `columns` (empty = every column, the old behaviour) and
`filter`. A source that cannot prune ignores both and returns everything, so the
change is strictly additive.
### The projection is DERIVED, not configured
`TableMapping::projected_columns()` reads the pipeline: a `vertex` step needs its
`id_column` + `properties`; an `edge`/`fk_edges` step needs its two endpoint
columns + `properties`. Every existing mapping therefore gets the narrowing with
no config change. It returns `None` — meaning *read everything* — when a `custom`
step is present, because an opaque registry transform may read any column and
narrowing its input would starve it. An explicit `columns:` list in the YAML
overrides the derivation.
### The predicate is backend-neutral
bifrost's core has no engine dependency, so the YAML filter is `mapping::RowFilter`
(`{ column, eq | in | ge/le }`), lowered to `skade::ScanFilter` at the single seam
inside the Iceberg source. A filter that names no form is REJECTED rather than
pushed down as a no-op.
```yaml
tables:
- table: people
mode: full
filter: { column: country, eq: SE } # partition / file pruning
pipeline:
- vertex: { label: Person, id_column: id, properties: [name] }
```
### Routing
| `columns` | `filter` | skade call |
|---|---|---|
| empty | `None` | `read()` — unchanged full scan |
| set | `None` | `read_columns(cols)` |
| any | `Some(f)` | `read_filtered(f, cols)` (empty cols = all columns) |
When a filter is present the source also costs the plan against the unfiltered
baseline (`plan_stats`) and logs `files_skipped` / `rows_planned`, so the
data-skipping win is observable rather than asserted. `IcebergSource::plan_stats`
exposes the same numbers to callers.
### Measured (`NORNIR_MACHINE=oden` — Threadripper PRO 3975WX, 32 physical cores, SMT off, 499GB RAM)
Bench arm `knut.iceberg_pushdown_read`: 1000000 rows, 11 columns, 4 data files
(one per country), projected to the 2 columns a `Person` mapping reads.
| arm | rows/s | vs full |
|---|--:|--:|
| full (pre-fix `read()`) | 2810017 | 1.00 |
| projected (2 of 11 columns) | 34023725 | **12.11** |
| projected + `country == SE` | 20454872 | **29.12** |
Plan-time skipping: `files_baseline = 4`, `files_planned = 1`, `files_skipped = 3`;
`rows_baseline_planned = 1000000`, `rows_planned = 250000`.
Correctness is gated separately: `crates/knut-bifrost/tests/iceberg_pushdown.rs`
asserts the projected read returns the SAME values for the columns it keeps as the
full read did, that the row count is unchanged, and that an unprojected/unfiltered
spec still reproduces the full-width read exactly.
### CDC is no longer full-width — CLOSED
The note that used to sit here said `SyncMode::Incremental` / `Cdc` "remain
full-width" because `skade::Table::read_changelog` takes no projection. That was
half right: `read_changelog` indeed takes none, but skade's `ChangelogBatch`
carries `project_columns`, and knut called neither. It does now — see
`cdc-design.md` §f4:
* **Spark path** — the `SELECT` over the changelog view (or over the Spark 4.2
`CHANGES` relation) is projected and filtered **in Spark**, instead of
`SELECT *`.
* **skade path** — `ChangelogBatch::project_columns` prunes the columns and
`cdc::post_filter` prunes the rows.
Metadata columns always survive (`_change_type` routes deletes), as does the
filter's own column; an unprojected/unfiltered spec reproduces the previous
statement exactly. Bench arm: `knut.cdc_pushdown_read`.
**Not the same claim as `Full`.** On the skade path this is a post-decode slice
and a post-scan row prune — it saves encode/transform/sink/memory, not parquet
I/O. The plan-time file skipping above (`files_skipped`) is a `Full`-only
number. Pushing projection into the manifest-delta scan itself is still
skade-side work.