nordisk/knut

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; the workflow in 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:

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

f1 — sources (TableSource)

src/source.rs. TableSpec { table, mode } (:11) names a table and its SyncMode. Four implementors, all behind the one trait:

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:

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 continues 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 PipelineSteps (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.

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 DELETEs 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 MERGEs 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).

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

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

tables:
  - table: people
    mode: full
    filter: { column: country, eq: SE }   # partition / file pruning
    pipeline:
      - vertex: { label: Person, id_column: id, properties: [name] }

Routing

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.

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.