nordisk/knut

Iceberg → FalkorDB — the complete guide

Class: guide · audience: someone who has not read the code and wants to get rows out of an Iceberg table and into a FalkorDB graph. This file is the single authority on which read paths exist, which transfer methods exist, and what must be running for each. Related: bifrost-guide.md (MappingSpec + listener), falkordb-guide.md (the loader API), falkordb-bulk-design.md (the GRAPH.BULK encoder).

(Lane letters below are that map's: A/B = Spark-free read + transfer, C = Spark Connect, D = remote clients.)

g0 — the one-minute answer

There are exactly two ways knut reads an Iceberg table, both in knut-bifrost, and both are library-only (no CLI):

There is no third path. knut-thund cannot read Iceberg (its native backend registers parquet and csv only — crates/knut-thund/src/backend/native.rs:852-867). knut-pipelines does not read Iceberg either; it only boots and drives Spark.

⚠ The bifrost CLI cannot do any of this. crates/knut-bifrost/src/bin/bifrost.rs:230 (binary gated on the cli feature) constructs FileSource::new(&files) and nothing else — there is no --warehouse, --iceberg or --spark flag anywhere in it. To stream Iceberg into a graph today you write a small Rust main. See §g5.

g1 — Spark-free: IcebergSource (map lanes A+B)

What must be running

Your own process, and a FalkorDB. That is the entire list. No JVM, no Spark, no Hive metastore, no REST catalog, no object store, no Kafka. skade reads Iceberg manifests in pure Rust, and the "catalog" is a local directory: <dir>/catalog.redb + <dir>/warehouse/. Opened at crates/knut-bifrost/src/source.rs:460 via skade::Warehouse::open.

FalkorDB, if you do not already have one:

podman run -d -p 6379:6379 localhost/knut-falkordb-valkey9:latest
# or knut's own image (FalkorDB module built from source on Valkey 9.1.0-trixie):
podman build -t knut-falkordb-valkey9 -f containers/falkordb-valkey9/Containerfile .

Two hard limits — read these before designing around this lane

1. The warehouse is a LOCAL FILESYSTEM PATH. IcebergSource calls only the local-path Warehouse::open. skade does have S3/object-store backends (open_with_store / open_with_object_store_config), but knut declares skade = { version = "0.5", default-features = false } (crates/knut-bifrost/Cargo.toml:120) and skade's default = [], so none of those backends are compiled in. An Iceberg table on S3/MinIO, or in a managed catalog, is not reachable on this lane — use the Spark Connect lane (§g2). (Superseded for bifrost watch: it now reads S3 with no catalog (--s3-root, skade-s3), any Iceberg REST catalog (--rest-uri, skade-rest, with --catalog aistor|polaris|unity|gravitino) and a remote skade-katalog (--grpc-uri, skade-grpc) — see §g5.1.)
2. redb is SINGLE-OPEN, and the source reopens the warehouse on every poll (source.rs:460, per fetch_new / latest_snapshot). If any other process — or an un-dropped Warehouse handle in your own process — holds the catalog, the round errors or blocks. This lane is a hand-off lane (writer commits and closes, poller opens), not a concurrent-writer lane. The integration test stages it exactly that way, dropping its writer handles before the listener reads (crates/knut-bifrost/tests/delta_iceberg_falkordb.rs:64-69, append_iceberg).

Read modes

Set by mode: in the MappingSpec YAML (see bifrost-guide.md §f1):

On the delta path: appended data files become untagged insert batches; equality-delete files become rows tagged _change_type=DELETE, which the listener routes to DETACH DELETE. An overwrite or position-delete commit sets needs_full_reload and the whole to snapshot is re-read, so the graph never silently goes stale. When the window is pure-append the _change_type column is stripped (source.rs:591 strip_change_type) so downstream sees plain inserts.

Reproduce it

cargo test -p knut-bifrost --features skade source::iceberg::tests

That unit test (source.rs:694+, reads_iceberg_table_and_tracks_snapshot) stages a real Iceberg table with skade and reads it back — no container, no network.

The end-to-end proof against a live database needs Podman:

systemctl --user start podman.socket
cargo test -p knut-bifrost --features skade --test delta_iceberg_falkordb -- --ignored

delta_iceberg_falkordb.rs:123 writes snapshot A (3 rows), asserts 3 Person nodes over a real GRAPH.QUERY, asserts an idle round ships nothing, then writes snapshot B (2 rows) and asserts the round shipped exactly 2 and the graph holds 5. It is #[ignore]d — it needs Podman and a FalkorDB image.

g2 — through Spark Connect: SparkConnectSource (map lane C)

What must be running

Your process, a Spark Connect server with the Iceberg runtime jar loaded, and FalkorDB.

The blessed pair is Spark 4.1.2 GA + iceberg-spark-runtime-4.1_2.13:1.11.0 (crates/knut-pipelines/src/server.rs:52-67, containers/spark-iceberg/Containerfile:16-24). Spark 4.2.0 exists as a named alternate but Iceberg publishes no iceberg-spark-runtime-4.2 yet, so 4.2 is for the SDP/pipelines path where no Iceberg jar is needed — do not use 4.2 for this lane.

podman build -t knut-spark-iceberg -f containers/spark-iceberg/Containerfile .
podman run -d -p 15002:15002 knut-spark-iceberg

This is the only lane that reaches a shared, remote or managed catalog (REST, Glue, Hive, Nessie) or a warehouse on object storage. You do not have to start a catalog server to use it: the shipped image configures a hadoop-type catalog lake on the container filesystem (containers/spark-iceberg/Containerfile:31-37, "zero external deps"), so the required service list is Spark + FalkorDB, two containers. A REST/Glue/Nessie catalog is optional and only needed if that is where your tables already live.

What it does

SparkConnectSource (source.rs:200) is a gRPC client that issues SQL strings:

• snapshot detection polls <table>.snapshots, after a REFRESH TABLE — necessary because Spark caches Iceberg metadata per session and would otherwise never see another writer's commits (source.rs:265);
• time-travel reads use VERSION AS OF, incremental uses EXCEPT ALL;
• true CDC calls the Iceberg stored procedure system.create_changelog_view(..., net_changes => true) (source.rs:251, net_changes at :254). net_changes collapses an update into an INSERT/DELETE pair and cancels an in-window insert+delete — exactly the split the graph wants.

Batches come back as workspace arrow-58 RecordBatch with no bridge. Both guards live in the module's #[cfg(test)] block: the never-called compile-time type guard _spark_collect_is_workspace_arrow58 (source.rs:360, the pinning binding at :366) and its runtime TypeId mirror (:376-378).

If the Spark server is up but missing the Iceberg jar, it fails loud at CREATE TABLE … USING iceberg — never a silent success. That is what crates/knut-bifrost/src/spark_demo.rs:130 run_iceberg_demo exists to surface.

g3 — ⚠ NO PUSHDOWN, on either lane

This is the single most important performance fact and it was previously undocumented.

Every Iceberg read in knut is a full-width scan. Neither lane expresses projection or predicate pushdown:

• The Spark-free lane calls skade's bare Table::read() — the whole snapshot, all columns — even though skade 0.5.1 exposes read_columns, read_filtered(ScanFilter{eq,is_in,range}) and plan_stats. knut wires none of them (grep read_columns\|read_filtered\|plan_stats crates/knut-bifrost/src/ returns nothing).
• The Spark lane issues SELECT *, so any pushdown is whatever Spark's own planner does server-side.

The only data reduction available today is the snapshot diff, which is file-level — not column, not predicate. Wiring read_columns / read_filtered into IcebergSource is a small, well-scoped improvement and is not yet done.

Also compiled out: skade's sql feature is off, so iceberg-datafusion-arrow58 is not in the build. knut has no SQL-over-Iceberg surface at all.

g4 — the transfer methods into FalkorDB (the menu)

Every path below writes Arrow. There is no CSV path, no GraphAr-file-on-disk detour, no in-memory-graph staging, and no falkordb_bulk_loader shell-out anywhere in the repo. GraphAr participates only as the _gar_id identity convention that edge endpoints key on.

Rates are from bench_bulk_vs_unwind (200k nodes, debug build, one FalkorDB container) — see falkordb-bulk-design.md. Treat the absolute nodes/s as lower bounds and the ratios (≈48× the parallel path, ≈108× the serial baseline) as the meaningful figures. A second, smaller run (60.6k nodes) in the same design doc measures ≈22× — the ratio is scale-dependent, so do not quote a single number.

The recommendation, and why

For an initial load: #1, and you do not choose it manually. choose_bulk(graph_is_empty) (crates/graphar-falkordb/src/loader.rs:789) picks it for you. It is ~48× the parallel UNWIND path because it is a pure-Rust binary encoder straight from Arrow onto FalkorDB's bulk wire protocol (crates/graphar-falkordb/src/bulk.rs) — one RESP command, no Cypher parsing, no Python, no CSV. It is CREATE-only and not transactional, which is precisely why it is gated on an empty graph.

For anything ongoing: #2. The bifrost FalkorDbSink (crates/knut-bifrost/src/sink/graph.rs:31) is the only idempotent write path in the entire repo. It emits UNWIND … MERGE (n:L {id: r.id}) SET n += r, double-MERGEs edge endpoints, and issues DETACH DELETE for CDC deletes — phase-ordered vertices → edges → deletes, with upserted_vertex_keys suppressing a delete of a key upserted in the same round so an UPDATE_BEFORE/UPDATE_AFTER pair does not destroy the node it just wrote. It also takes the GRAPH.BULK fast lane automatically when the target label is empty and the mutation is vertex-only (sink.rs:323 bulk_lane_eligible), so you get #1's speed on first fill and correctness on every round after. On crash-replay the label is now populated, so it falls back to MERGE — replay-safe.

⚠ Nothing in graphar-falkordb is idempotent. Methods #1, #3 and #4 all CREATE. Re-running a load duplicates every node. Only create_index is idempotent. If you need upsert semantics, you need the bifrost sink (#2).

The remote-path performance cliff

Flight SQL DoPut (do_put_statement_update, crates/graphar-flight/src/service.rs:375) calls the serial insert_vertices (:404) / insert_edges (:412) — not the _par siblings, and not GRAPH.BULK. A Python client bulk-loading a lake table is therefore ~100× slower than the same load from in-process Rust. Routing DoPut through insert_*_par, or through bulk_insert_streamed when the target label is empty, is an obvious unclaimed optimisation.

g5 — the whole thing, end to end (Spark-free)

The one-call entry point. This is real streaming: a poll loop with a persisted snapshot cursor and idempotent MERGE.

// Cargo.toml: knut-bifrost = { version = "*", features = ["skade"] }
use std::time::Duration;

knut_bifrost::listener::watch_iceberg_delta(
    spec,                                   // MappingSpec::load("mapping.yml")?
    "/var/lib/lake",                        // the skade warehouse DIRECTORY
    "redis://127.0.0.1:6379",               // FalkorDB
    "social",                               // graph name
    "/var/lib/bifrost-state.json",          // the snapshot cursor
    Duration::from_secs(15),
    |round| { println!("{round:?}"); true },  // return false to stop
).await?;

crates/knut-bifrost/src/listener.rs:372. Equivalent long form, if you want to swap the source for SparkConnectSource::connect("sc://host:15002") (the Spark lane, §g2) without changing anything else — Listener is generic over TableSource:

let mut listener = Listener::new(spec, registry, IcebergSource::new("/var/lib/lake"),
                                 FalkorDbSink::connect(url, "social").await?,
                                 SyncState::open("./state.json")?);
listener.run(Duration::from_secs(15), |r| !r.is_failure()).await?;

This is an async fn — your process owns the loop. The cursor is durable, so a restart resumes without re-ingesting, but nothing restarts it for you — unless you run it as the bifrost falkordb service (§g5.1).

g5.1 — bifrost watch as a service: bifrost falkordb

The same watch flags, installed through skidbladnir as a systemd unit or rendered as a Kubernetes StatefulSet + Secret + cursor volume. Feature falkordb-service (no Doris sink); service includes it.

bifrost falkordb unit      <watch flags>                 # print the unit install would write
bifrost falkordb install   <watch flags> [--system]      # write, enable, start
bifrost falkordb status|start|stop [--name N]
bifrost falkordb uninstall [--purge]                     # --purge also drops the cursor
bifrost falkordb k8s       <watch flags> --image I [--namespace NS] --out DIR

Which lake (exactly one): --warehouse DIR · --s3-root s3://… · --rest-uri URL (--catalog rest|aistor|polaris|unity|gravitino, --rest-warehouse, --rest-scope, --rest-oauth2-uri, --rest-header, --vended-credentials, --gravitino-user) · --grpc-uri URL (--grpc-ca/--grpc-cert/--grpc-key for an https:// front; skade-katalog serve --grpc-tls-cert/--grpc-tls-key [--grpc-client-ca] is the other end). A vendor catalog the build lacks fails at run time naming its feature.

Secrets are environment only, copied into the unit's secret channel or the k8s Secret, never onto a command line: MINIO_ACCESS_KEY, MINIO_SECRET_KEY, BIFROST_S3_CA_PEM, BIFROST_FALKOR_URL (a FalkorDB URL with a password must come this way), BIFROST_CATALOG_CREDENTIAL (OAuth2 id:secret), BIFROST_CATALOG_TOKEN (bearer).

Kubernetes: skidbladnir mounts no ConfigMap, so the mapping file and any PEMs must be inside the image, named by absolute path (k8s refuses relative ones). The image is containers/bifrost/Containerfile.

Pick the smallest build (Build Thing rows, crates/knut-build-thing/src/recipe.rs):

dwarves buildthing knut --variant bifrost-falkordb-polaris

"Can I submit it and walk away, like spark-submit?"

Not today, on any path, and the gap is smaller than it looks:

• The bifrost CLI is the headless detachable driver — JSON logs, graceful shutdown, Prometheus /metrics (that endpoint is behind the metrics-http feature; the binary itself behind cli) — but it is hard-wired to FileSource (bin/bifrost.rs:230). The entire gap is one source constructor and a --warehouse flag. It is not architectural.
• SDP (Spark Declarative Pipelines) is not the answer. SDP pipelines are authored in Python or SQL only — there is no Rust and no plugin ABI, so SDP will never run your Rust function. And spark-pipelines run "monitors its progress until it completes": it is built on spark-submit and has no documented detached mode in 4.1 or 4.2. It gives declarative dataset definitions with dependency ordering; it does not give detachment.
• jera is the constellation's real submit-and-detach substrate — a durable redb job ledger with a gRPC agent transport. But its WorkPayload is Process | Container | Vm: you ship a spec naming an already-built binary or image, never a Rust function. Once the CLI gap above is closed, packaging bifrost as a container and dispatching it as a WorkPayload::Container is spark-submit's ergonomics, with a better job ledger.
• "Insert a Rust function" does exist — Transform (src/transform.rs:26), plus wasm (src/wasm.rs) and Arrow-Flight (src/flight.rs) plugin hosts — but all three execute in the bifrost process, i.e. in the client. The wasm host is the closest thing to shippable sandboxed user code, and it already speaks Arrow IPC in/out.

g6 — the reverse direction (graph → Iceberg)

For completeness, the mirror exists and is benched:

cargo run -p graphar-flight --features skade --example export-iceberg -- \
    redis://127.0.0.1:6379 mygraph \
    "MATCH (n:Person) RETURN n._gar_id AS id, n.name AS name" \
    /tmp/knut_warehouse people

export_cypher_to_iceberg (crates/graphar-flight/src/export.rs:218) uses skade's zero-copy Arrow bridge (to_skade_batch, :327) + fast_append: ≈5.37M rows/s versus ≈2.75M through the parquet temp-file bridge.