# 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`](bifrost-guide.md)
> (MappingSpec + listener), [`falkordb-guide.md`](falkordb-guide.md) (the loader API),
> [`falkordb-bulk-design.md`](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):
| lane | feature | what must be RUNNING | reaches |
|---|---|---|---|
| **A/B — Spark-free** (`IcebergSource`) | `skade` | **your process + FalkorDB.** Nothing else. | a **local directory** warehouse only |
| **C — Spark Connect** (`SparkConnectSource`) | `spark` | your process + **a Spark 4.1.2 JVM** + FalkorDB | any REST / Glue / Hive / Nessie / S3 catalog |
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:
```bash
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`](bifrost-guide.md) §f1):
| `mode` | what it reads | code |
|---|---|---|
| `full` | the whole table at the new snapshot, every time | `source.rs:609` (`impl TableSource`), `table.read()` at `:639` |
| `incremental` / `cdc` | manifest-level snapshot **diff** | `source.rs:505` `delta_from_table` → `skade::Table::read_changelog` |
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
```bash
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:
```bash
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**.
```bash
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.
| # | method | rate | idempotent? | use when |
|---|---|---|---|---|
| 1 | **`GRAPH.BULK`** — `bulk_insert_streamed` | **~2.69M nodes/s** | ✗ CREATE-only | **initial load into an EMPTY graph** |
| 2 | **bifrost `FalkorDbSink`** (MERGE) | escalates to #1 when the label is empty | **✓ the only one** | **streaming / incremental / CDC** |
| 3 | parallel `UNWIND` — `insert_*_par` | ~56k nodes/s | ✗ | manual loads onto a non-empty graph |
| 4 | serial `UNWIND` — `insert_vertices` | ~25k nodes/s | ✗ | deterministic single-link ordering |
| 5 | Arrow Flight SQL `DoPut` | ~25k nodes/s (lands on #4) | ✗ | remote / non-Rust clients only |
| 6 | Spark Connect → sink | — | ✓ (uses #2) | remote catalog, or full UPDATE_BEFORE/AFTER |
Rates are from `bench_bulk_vs_unwind` (200k nodes, **debug** build, one FalkorDB container)
— see [`falkordb-bulk-design.md`](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.
```rust
// 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`:
```rust
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.
```text
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`):
| row | carries |
|---|---|
| `bifrost-falkordb` | watch + service; warehouse, S3, generic REST |
| `bifrost-falkordb-aistor` / `-polaris` / `-unity` / `-gravitino` | + that ONE vendor catalog |
| `bifrost-falkordb-grpc` | + the skade-katalog gRPC client (tonic) |
| `bifrost-full` | every sink (Doris too) and every catalog |
```text
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:
```bash
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.