nordisk/knut

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'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). 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 (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). 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). 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_exchangees 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 (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).