<!-- nornir:class: unsorted -->
# Plan — proposed functionality
What knut should grow next, roughly ordered by leverage. Items marked ⭐ are
the ones that change what knut *is*, not just what it has.
## ⭐ 1. Airflow 3 asset events (AIP-82) — the file/table listener as a citizen of the DAG world ✓ DONE (2026-06-10)
> Shipped: `knut_bifrost::airflow` (`AssetEvent`, `AssetEmitter`,
> `round_to_events`, `emit_round`, and the feature-gated `AirflowEmitter`
> POSTing to `/api/v2/assets/events`) + `knut_bifrost::source::FileSource`, a
> `TableSource` over a watched directory whose `TableState::done_files` ledger
> is the "which files are done" bookkeeping. Unit-tested; OSM e2e in
> `tests/integration_osm_filesource.rs`. Detail below = future depth.
Airflow 3's event-driven scheduling (AIP-82, `AssetWatcher` +
`BaseEventTrigger`) lets a DAG run when an external event lands, not on a
cron. Bifröst already has the two halves AIP-82 wants:
- **Emit**: after every successful sync round, POST an asset event to
Airflow's API (`/assets/events`) or push to the Common Message Queue —
"table X is now in the graph as of snapshot S". Downstream DAGs
(analytics, exports, alerts) schedule on that asset.
- **Listen**: a `FileSource` implementing bifröst's `TableSource` over a
watched directory / S3 prefix (GraphAr drops, parquet dumps), with
`SyncState` as the "which files are done" ledger — exactly the bookkeeping
AIP-82's poll-based triggers expect. New file → sync round → asset event.
Concretely: a small `knut-airflow` crate (or a `bifrost --emit-airflow` flag)
plus a documented `AssetWatcher` recipe on the Airflow side.
## ⭐ 2. Power BI over Flight SQL (ADBC) ✓ DONE (2026-06-10)
> Shipped in `graphar-flight`: the metadata surface (`GetCatalogs`,
> `GetDbSchemas`, `GetTables`, `GetTableTypes`, `GetSqlInfo`, prepared
> statements) with registered queries surfaced as tables (named via
> `SchemaRegistry::register_named`) and `SELECT * FROM <table>` rewritten to
> the underlying Cypher. Auth covers what Windows BI clients use: bearer,
> Basic + Flight handshake (token issuance), and mutual TLS — see
> `graphar_flight::auth::AuthConfig` and `ServerOptions`. Unit-tested; live
> path in `tests/integration_flight_sql.rs` via the stock `FlightSqlServiceClient`.
Power BI switched new connections to **ADBC by default in February 2026**
(full transition planned July 2026), and Dremio already ships an
ADBC/Flight-SQL-based Power BI connector — the wire knut already speaks.
To get knut dashboards in Power BI:
- Implement the Flight SQL **metadata surface** (`GetTables`, `GetDbSchemas`,
`GetSqlInfo`, prepared statements) in graphar-flight so generic
Flight SQL ODBC/ADBC drivers can introspect it. Registered Cypher queries
appear as "tables"; auto-schema covers ad-hoc ones.
- Document the path: Power BI → Arrow Flight SQL ODBC driver (Dremio's
generic build) or `adbc_driver_flightsql` → `graphar-flight` with TLS +
bearer.
## 3. Iceberg source (drop Spark for append-only tables) ✓ DONE (2026-06-11)
> Shipped: `IcebergSource` (feature `skade`) — a `TableSource` that reads an
> Iceberg table directly through [`skade`](https://crates.io/crates/skade)'s
> embedded catalog, **no Spark**. Each poll compares the table's current
> snapshot id (`inner().metadata().current_snapshot()`) to the recorded one and
> re-reads on change (idempotent MERGE makes that safe). skade and knut now
> share one **Arrow 58** tree, so the seam (`bridge_57_to_56`, mirroring
> graphar-flight's `bridge_56_to_57`) is a plain identity `clone()`; it began as
> a **zero-copy** Arrow C Data Interface transmute when skade was on 57 and knut
> on 56 (see [`reasoning.md`](reasoning.md)). From crates.io
> (`skade = { version = "0.5", default-features = false }`), no path coupling —
> and, since skade 0.5.0 published the equality-delete surface, no root
> `[patch.crates-io]` sibling override either.
> Unit-tested (no container) in `source::iceberg::tests`. Embedded catalog, not
> REST — but Spark-free, which was the point; a REST catalog could swap in
> behind the same `Warehouse`/iceberg-rust later.
`iceberg-rust` can read snapshots + parquet data files directly from the
catalog; for append-only tables bifröst wouldn't need a Spark cluster at all.
Same `TableSource` trait, much smaller footprint.
## 4. Changelog-view incremental reads (updates + deletes) ✓ DONE (2026-06-11)
> Shipped: a new `SyncMode::Cdc`. `SparkConnectSource` reads Iceberg's
> `system.create_changelog_view` (carrying `_change_type`) instead of
> `EXCEPT ALL`; `transform::partition_change_rows` splits each round into
> upserts (INSERT/UPDATE_AFTER) and deletes (DELETE/UPDATE_BEFORE), strips the
> sentinel column, and `GraphMutation.deletes` (a `VertexDelete` set) drives
> `MATCH (n:Label {id}) DETACH DELETE n` in the sink (run *after* upserts so a
> same-round upsert+delete resolves to gone). `SinkReport.deleted` + a
> `vertices_deleted` log event. Caveat: edge-only changelog deletes are out of
> scope (a vertex DETACH DELETE already removes its incident edges).
`EXCEPT ALL` only sees appends. Iceberg's
`system.create_changelog_view` exposes INSERT/UPDATE/DELETE row-level diffs;
map DELETE → `MATCH … DETACH DELETE` in the sink and bifröst becomes a real
CDC consumer.
## 5. Graph → lakehouse round trip ✓ DONE (2026-06-11)
> Shipped (part 1): `graphar_flight::export::export_cypher` writes a Cypher
> query's rows as a GraphAr chunk file (FalkorDB → Arrow → lakehouse). Runnable
> via `cargo run -p graphar-flight --example export`; round-trip in
> `tests/integration_export.rs`.
>
> Shipped (part 2, feature `skade`): `export_cypher_to_iceberg` appends the rows
> straight into an **Iceberg table** via [`skade`](https://crates.io/crates/skade)
> (the Iceberg read/write stack over an embedded catalog). skade and knut now
> share one **Arrow 58** tree, so `bridge_56_to_57` is a plain identity `clone()`
> (an `Arc` bump — no temp file, no serialization); it was a **zero-copy** Arrow
> C Data Interface transmute back when skade was on 57 and knut on 56 (see
> [`reasoning.md`](reasoning.md)). `append_arrow_to_iceberg_via_parquet` keeps the old
> temp-parquet path as the baseline the bridge beats (≈2× — see the two
> `knut.skade_iceberg_export*` benchers in [`benching-benchmarks.md`](benching-benchmarks.md)). Runnable
> via `--features skade --example export-iceberg`; the bridge + write/read-back is
> unit-tested (no container) in `export::skade_tests`.
>
> Shipped (part 3): `export_cypher_chunked(exec, cypher, dir, FileType,
> chunk_rows)` writes the result as multiple GraphAr chunk files (`chunk0.<ext>`,
> `chunk1.<ext>`, …) of ≤ `chunk_rows` rows each, so write-side/encoder memory
> and per-file size stay bounded by `chunk_rows` for huge results; `ExportReport`
> gained an additive `chunks: Vec<PathBuf>`. **Slice-and-write, not pull-stream**:
> FalkorDB's `GRAPH.QUERY` is one RESP reply with no server-side cursor, so the
> full result is fetched once and then **zero-copy-sliced** (`RecordBatch::slice`
> shares buffers) — true streaming from FalkorDB would need a cursor API RESP
> doesn't expose. Round-tripped (no container) in `export::chunk_tests`: N files
> for `chunk_rows < total`, each bounded, total rows preserved.
The reverse arrow (now done): `export_cypher` writes a query's rows back as a
GraphAr file and `export_cypher_to_iceberg` straight into an Iceberg table, so
knut covers both directions of the columnar ↔ graph bridge.
## 6. WASM / Flight transform plugins ✓ DONE (2026-06-11)
> Shipped: `knut_bifrost::wasm::WasmTransform` (feature `wasm`) — a `Transform`
> whose logic is a sandboxed wasm module run by the **pure-Rust `wasmi`**
> interpreter (no C/JIT). ABI: the guest exports `memory` + `alloc(len)->ptr` +
> `transform(ptr,len)->packed(ptr,len)`; the host serializes input batches to
> **Arrow IPC**, hands them in, reads the IPC result back, and turns each output
> batch into a `VertexSet` — same `GraphMutation` contract, now polyglot +
> out-of-line. `from_bytes`/`from_file`; tested with an inline WAT identity guest
> (rows round-trip through the IPC bridge).
>
> **Real `wasm32` guest example shipped (2026-06-11): `crates/knut-bifrost/wasm-guest/`.**
> A standalone `cdylib` crate (NOT a workspace member — listed in
> `[workspace] exclude`, carries its own empty `[workspace]` table) that compiles
> to `wasm32-unknown-unknown` and implements a **real arrow-on-wasm transform**:
> read the input Arrow IPC stream (`arrow_ipc::reader::StreamReader`), **keep only
> rows where `_gar_id` is even** (`arrow_select::filter::filter_record_batch`),
> write the result back as an IPC stream (`StreamWriter`) into a leaked buffer,
> and return the packed `(out_ptr<<32)|out_len`. Exports `memory` + `alloc`
> (`Vec`-leak) + `transform`, all `#[no_mangle] pub extern "C"`. **Arrow 58
> compiles to `wasm32-unknown-unknown` out of the box — no patches/shims/feature
> gymnastics** (arrow-array/-schema/-buffer/-data/-ipc/-select + flatbuffers,
> chrono, half, ahash, getrandom 0.2, num-\* all build clean for bare-unknown).
> Build: `cargo build -p knut-bifrost-wasm-guest --target wasm32-unknown-unknown
> --release --manifest-path crates/knut-bifrost/wasm-guest/Cargo.toml` →
> `wasm-guest/target/wasm32-unknown-unknown/release/knut_bifrost_wasm_guest.wasm`
> (~1.2 MB, `opt-level="z"`+lto+`panic="abort"`). An `#[ignore]` integration test
> (`tests/wasm_guest.rs`, feature `wasm`) loads the artifact into a
> `WasmTransform` and asserts only even-`_gar_id` rows survive — run with
> `cargo test -p knut-bifrost --features wasm --test wasm_guest -- --ignored`
> (ignored by default since it needs the pre-built `.wasm`). See
> `wasm-guest/README.md`.
>
> **Flight half shipped (2026-06-11): `knut_bifrost::flight::FlightTransform`
> (feature `flight`).** A `Transform` that **proxies** the transform to an
> **external Arrow-Flight service over `DoExchange`** (the bidirectional Flight
> stream): `connect(endpoint, label, id_column)` opens a `tonic` channel; each
> `apply` encodes the input `&[RecordBatch]` as a Flight data stream
> (`FlightDataEncoderBuilder`), `do_exchange`es it, decodes the returned batches
> (`FlightRecordBatchStream`), and turns each into a `VertexSet{label,id_column}`
> — the **same `GraphMutation` contract as `WasmTransform`, but out-of-process
> and polyglot** (the service can be any language with an Arrow Flight server).
> `Transform::apply` is sync, so it bridges to the async `FlightClient` with a
> small ambient-runtime-aware `block_on` (hands the future to a dedicated
> current-thread runtime when already inside a Tokio runtime, so a
> current-thread worker can't deadlock). All transport errors map to
> `BifrostError::Other` prefixed `flight:`. Pulls `arrow-flight` + `tonic`
> (workspace versions, same 56/0.13 tree as graphar-flight) only under the
> off-by-default `flight` feature — the default build's dep tree is unchanged.
> Tested **with no external infra**: an in-process echo `DoExchange`
> `FlightService` is bound on an ephemeral port (`TcpIncoming`) on a background
> task, a `FlightTransform` points at it, and `apply` over a sample batch is
> asserted to round-trip the rows into `GraphMutation.vertices`
> (`flight::tests`, gated on `feature = "flight"`). **#6 now fully done.**
The fabric's escape hatch is compile-time Rust today. A WASM host (transform
= module taking/returning Arrow IPC bytes) or an external Arrow Flight
transform service would allow out-of-process, polyglot transforms with the
same `GraphMutation` contract.
## 7. Operational hardening
- ~~`bifrost` binary~~ (done): `bifrost run` with structured JSON logs +
graceful Ctrl-C shutdown (finishes the round + persists, then exits).
- ~~Metrics + backoff~~ (done, 2026-06-11): `knut_bifrost::metrics` — a
cumulative `Metrics` accumulator (rounds_total, rounds_failed, tables_synced,
rows_ingested, vertices_written, edges_written, deletes, assets_emitted) folded
after every round, emitted as a structured `metrics` tracing event (target
`metrics`) and exposed via `Listener::metrics()`; `render_prometheus(&Metrics)`
is a pure function (no HTTP-server dep — the CLI `--metrics-file` writes the
text on shutdown). **Backoff**:
a fully-failed / no-progress round (`RoundReport::is_failure` — idle ≠ failure)
grows the inter-round sleep `interval → 2× → … → max` (default `interval*10`,
`--max-backoff` to override) and the first round that makes progress resets it;
the happy-path cadence is unchanged. Wired into `Listener::run` (additive —
same signature) + `run_with_backoff(Backoff, …)` and the CLI poll loop. Unit
tests cover backoff growth/reset/clamp + metrics accumulation/Prometheus
rendering + end-to-end through `sync_once`.
- ~~Live `/metrics` HTTP endpoint~~ (done, 2026-06-11): `knut_bifrost::metrics_http`
(`MetricsServer`/`MetricsHandle`, feature **`metrics-http`**) serves
`render_prometheus(&Metrics)` over HTTP at `/metrics` so a Prometheus scraper
can **pull** the counters *live*, not just read the file written on shutdown.
A tiny, pure-Rust, blocking `tiny_http` server runs on a background thread (no
async runtime); the poll loop publishes the cumulative `Metrics` into the
server's shared `Arc<Mutex<Metrics>>` snapshot after each round, so every
scrape renders a consistent, current snapshot. `GET /metrics` → `200` +
`text/plain; version=0.0.4`; unknown paths → `404`. CLI flag `--metrics-addr
<ip:port>` (e.g. `127.0.0.1:9090`) starts it; absent → no server (unchanged
behavior). `tiny_http` is the only new dependency and only under
`metrics-http` (the default/core build is byte-for-byte dependency-unchanged);
`cli` does **not** imply it. Unit-tested in `metrics_http::tests`: an
ephemeral-port server + a dependency-free `TcpStream` GET asserting the body
carries `bifrost_rounds_total`, plus live-publish and 404 paths.
- ~~Valkey-backed `SyncState`~~ (done, 2026-06-11): `SyncState::open_valkey`
keeps the ledger in a Valkey 9 hash so it survives multi-instance (the
Redis-API/RESP store FalkorDB already runs on). File backend remains default.
- Flight SQL: ~~mTLS~~ (done), ~~token rotation~~ (done, 2026-06-11),
~~per-query authz~~ (done, 2026-06-11). **Token rotation**: `AuthConfig`
now holds a *set* of accepted bearer tokens (`with_bearer_tokens` /
`add_bearer_token`, plus `with_named_bearer_tokens` to map several tokens to
one authz identity) — any listed token authenticates, the first is the one
`do_handshake` issues, so a secret rotates with zero downtime (add new →
migrate → drop old, both valid during the overlap). `with_bearer_token(one)`
is preserved as a one-element set; `ServerOptions` gained `bearer_tokens`
alongside the unchanged `bearer_token`. **Per-query authz**: the interceptor
resolves an `Identity` (`Token`/`Basic`/`Anonymous`) and stashes it as a
request extension; after a query resolves to its registered name the service
consults `AuthConfig.authorizer` — an `Authorizer` enum (`AllowAll` default,
`AllowList(identity → {query names})` with a `"*"` wildcard, or a
`Custom(Fn)`) — and a denial is `permission_denied`. Enforced at every Cypher
entry point (`get_flight_info_statement`, `do_get_statement`, both prepared
read paths, `do_action_create_prepared_statement`). Additive on
`AuthConfig`/`ServerOptions`; the open-server and existing
single-token/Basic/mTLS paths are unchanged (single token = one-element set,
default `AllowAll` = no authz check). Unit-tested: multi-token acceptance +
stale-token rejection (rotation) and the allow/deny decision (a token allowed
`q1`, denied `q2`) at both the `AuthConfig` and the interceptor.
- ChunkStore: ~~retries/backoff config~~ (done, 2026-06-11): remote ops use
`object_store`'s native `RetryConfig`, fronted by a typed `RetryOptions`
(`max_retries` / `init_backoff` / `max_backoff` / `retry_timeout`).
`from_url` / `s3_compatible` keep the library's default policy (behavior
unchanged); `*_with_retry(.., RetryOptions)` overrides, `RetryOptions::none()`
opts out.
- ChunkStore: ~~multipart upload for big chunks~~ (done, 2026-06-11): writes
whose encoded size exceeds `MultipartOptions::threshold` (default 16 MiB)
stream to the backend via `object_store::put_multipart` in `part_size`-byte
parts (default 8 MiB, ≥ the 5 MiB S3/GCS/R2 minimum); smaller writes keep the
single `put`. Both `write_chunk` and `put_bytes` route through it. Tune via
`ChunkStore::with_multipart(MultipartOptions::new(threshold, part_size))`;
defaults keep all local / in-memory traffic on single-`put`. Parts upload
sequentially and a failure `abort()`s the upload so S3/GCS don't leak parts.
## 8. Spec completeness
- GraphAr: validate against the official gar/v1 testdata corpus — **done**,
2026-06-11. Two-tier, offline-first: an always-on **structural validator**
(`graphar::validate` — `validate_graph_dir` + `GraphInfo::validate`, returning
`Vec<ValidationIssue>`) checks a written graph dir against the `gar/v1` rules
(info-file fields well-typed, `version == gar/v1`, chunk sizes > 0, every
declared property group / adjacency-list / offset directory present, chunk &
partition files named + numbered per spec — partitions may be sparse, chunks
within a partition gapless, ≤ 1 primary, each property in exactly one group),
exercised by a **round-trip conformance test** (build with
`VerticesBuilder`/`EdgesBuilder` → dump → assert exact spec paths +
`validate()` clean). Plus an opt-in `#[ignore]` **corpus test** gated on
`GRAPHAR_TESTDATA` pointing at a local `apache/incubator-graphar-testing`
checkout (no network; `ValidateOptions::corpus()` for the corpus's
extensionless chunk files). Verified during development against a real
checkout: 12 gar/v1 graphs validated cleanly, 4 skipped (the non-`gar/v1`
`java/` subtree + one `list<float>` parameterized-type graph knut's `DataType`
doesn't model). Reading the corpus also required making `GraphInfo.prefix`
`#[serde(default)]` (the upstream `*.graph.yml` omit it). (`Date`/
`Timestamp`/`List` cypher rendering — **done**, 2026-06-11. Edge property
groups in bifröst mappings — **done**, 2026-06-11: `edge`/`fk_edges` steps
take `properties: [...]` + optional `rename`, projected alongside `src`/`dst`
into the `EdgeSet` and folded into the relationship via `SET e += r.props`.)
- Label/property-graph niceties: composite MERGE keys, multi-label nodes —
**done**, 2026-06-11 (`graphar-falkordb`). `FalkorDbLoader::insert_vertices_multi`
emits `CREATE (n:`A`:`B`:`C`)`; `insert_edges_keyed(src, type, dst, &src_keys,
&dst_keys, &batches)` MATCHes each endpoint on N key props (`MATCH (a:`L`
{`k1`: …, `k2`: …})`), the carrier columns generalizing from `src`/`dst` to
the named key columns. Additive: `insert_vertices`/`insert_edges` delegate
(one-label slice / `["_gar_id"]`) and the single-key/single-label wire shape
is byte-identical (legacy `_src`/`_dst` carriers, plain `` `Label` ``). Pure
query-builder unit tests assert the multi-label CREATE and composite-key MERGE
Cypher.
---
## Handoff state (2026-06-09, end of session)
- Everything above "proposed" in the README roadmap is **done and tested**:
65 unit tests + live podman matrix (FalkorDB-on-Valkey9, Spark 4.2-preview5
+ Iceberg 1.11.0 TOP GUN e2e, SeaweedFS, RustFS) all green.
- **Working tree is uncommitted** — ~7.6k insertions across 20 modified files
+ new: crates/knut-bifrost, xtask, containers/, docs/bifrost-design.md,
.nornir/, integration tests. Commit in logical chunks (formats → storage →
flight auth → bifrost → containers/tests → benches/docs).
- Local podman images built: `knut-falkordb-valkey9` (valkey 9.1.0 + module
compiled in-container, redis-compat gate patched), `knut-spark-iceberg`.
- Gotchas are recorded in `.nornir/design.md` ("Decisions worth remembering").
## Handoff state (2026-06-10)
- ⭐1 (Airflow AIP-82) and ⭐2 (Power BI / Flight SQL) above are **done** — see
the `✓ DONE` notes under each. Plus full Flight SQL auth (bearer + Basic
handshake + mTLS) and a `knut.file_source_scan` nornir bench.
- All unit + doc tests green (`cargo test --workspace`); two new `#[ignore]`
integration tests compile: `graphar-flight/tests/integration_flight_sql.rs`
(Power BI path) and `knut-bifrost/tests/integration_osm_filesource.rs`
(OSM `relations.parquet` slice → FileSource → FalkorDB; needs
`KNUT_OSM_RELATIONS` + the FalkorDB image).
- New feature flags: `knut-bifrost/airflow` (reqwest emitter),
`graphar-flight` gained `base64` dep for Basic auth.
- DONE: the `knut-pipelines` crate — a viewer for Spark Declarative Pipelines
**and** plain Spark Connect jobs. Layers: `model` (DAG) + `progress`
(`ExecutionProgress`/`StageProgress` job progress + unified `ViewerState`),
`client` (`grpc`: pipeline events **and** job progress via `run_updates` /
`execute_sql`), `viewer` (text/JSON + `dump_state`), `gui` (`gui` feature:
egui + **egui-snarl** live node-graph DAG, progress bars, event log), and the
`knutpipe` binary (`show`/`watch`/`gui`). Features all off by default
(`grpc`, `gui`, `live = gui+grpc`); default build is pure model + CLI.
egui/eframe pinned to 0.28 (egui-snarl 0.4 compat). Verified: default 20
tests, grpc 26, gui 22, all four feature combos warning-free; GUI window
smoke-tested live; full workspace green. See `docs/pipelines-viewer.md`
("Autonomy decisions" records calls made while the user was away).
## Handoff state (2026-06-11) — current
Everything is **committed and pushed** (`codeberg.org:nordisk/knut`, branch
`main`). All §-marked roadmap items are done except where noted:
- **#1 Airflow AIP-82**, **#2 Power BI / Flight SQL** + full auth (bearer /
Basic-handshake / mTLS), **#4 changelog CDC**, **#5 graph→lakehouse** (GraphAr
file *and* Iceberg table via `skade`), **#3 Spark-free Iceberg source** via
`skade`'s `IcebergSource`, **#7** Valkey shared `SyncState` + graceful
shutdown, **#8** Date/Timestamp/List Cypher rendering — all shipped.
- **Viewer**: `knutpipe` does `show`/`watch`/`gui`/**`pipeline`** (define + run
an SDP, terminal or egui-snarl GUI), verified live against Spark 4.2.
- **skade** is a crates.io dep (`skade = { version = "0.5", default-features =
false }`, off-by-default `skade` feature in graphar-flight + knut-bifrost) —
no path coupling to the sibling repo, and no root `[patch.crates-io]`
override: the equality-delete surface the patch existed for shipped in 0.5.0.
- **Perf**: the whole audit (H1–H12, M1–M5) is **done** — see [`perf.md`](perf.md)
(every item `[x]`) + the `knut.cypher_render` bench. Includes the Flight SQL
no-double-execution schema cache (H9) and the viewer idle-cost fixes (H11/H12).
- `cargo test --workspace` green (21 test-result groups); every feature combo
builds warning-free. Docs rendered to `docs/{README.pdf, book.pdf, book.html}`.
- **Open**: perf backlog; WASM/Flight transform plugins; a REST Iceberg catalog
behind skade's `Warehouse`; Iceberg-table CDC export; switching bulk writes to
skade's `ingest_pipelined` (~12% faster, available in skade ≥ 0.4.7).