# Arrow → FalkorDB bulk load — guide
> **Class:** `guide` · audience: a techie bulk-loading Arrow batches into FalkorDB.
> The *what* is in [`falkordb-design.md`](falkordb-design.md); the *why* in [`falkordb-reasoning.md`](falkordb-reasoning.md);
> the fastest path, `GRAPH.BULK`, is in [`falkordb-bulk-design.md`](falkordb-bulk-design.md).
> This file is the API and the load workflow.
> ⚠ **Nothing in this crate is idempotent.** Every method below `CREATE`s — the `UNWIND`
> vertex path, the `UNWIND` edge path, and `GRAPH.BULK` alike. **Re-running a load
> duplicates every node.** Only `create_index` is safe to repeat. If you need upsert /
> replay-safe semantics you want `knut-bifrost`'s `FalkorDbSink`, which is the only `MERGE`
> path in the repo — see [`iceberg-falkordb-guide.md`](iceberg-falkordb-guide.md) §g4.
## f0 — chunk to graph in one round trip (the whole workflow)
1. **Connect** to a FalkorDB graph: `FalkorDbLoader::connect(url, graph).await`.
2. **Load vertices**, then edges: `insert_vertices(label, &batches)`, then
`insert_edges(src, type, dst, &batches)`. Vertices ship one `UNWIND … CREATE` per
window; edge windows ride a `CYPHER rows=[…]` **query parameter** into a constant
`UNWIND $rows … MATCH` endpoints `… CREATE` body, so FalkorDB parses/plans the edge
query once and the plan cache serves every later window.
3. Nodes get a `_gar_id` property (absolute row index) automatically — that's the key edges
resolve on.
**The edge lane indexes its own endpoint keys** before its first window: one idempotent
`create_index` per (label, key property) — FalkorDB coalesces per-property creates into
the label's one index, and a composite `MATCH (a:L {k1:…, k2:…})` plans a
`Node By Index Scan` off it. A manual `create_index(label, "_gar_id")` between vertices
and edges is still fine (idempotent) but no longer load-bearing. Without an index the
plan is a per-row label scan — measured live 2026-10-20 at **~17 edges/s**.
**If the graph is empty, skip all of this and use `GRAPH.BULK` (§f2b) — it is ≈48× faster.**
## f1 — the loader API
```rust
use graphar_falkordb::FalkorDbLoader;
let mut loader = FalkorDbLoader::connect("redis://127.0.0.1:6379", "social").await?;
// optional: override the 512-row default batch:
let mut loader = loader.with_batch_size(1024)?;
let n = loader.insert_vertices("person", &vertex_batches).await?;
// insert_edges guarantees the (label, key) endpoint indexes itself.
loader.insert_edges("person", "KNOWS", "person", &edge_batches).await?;
loader.run_query("MATCH (n:person) RETURN count(n)").await?; // arbitrary Cypher
```
`connect(url, graph)` opens a multiplexed async Redis connection to one named graph.
`with_batch_size(n)` overrides `DEFAULT_BATCH_SIZE` (512); `n == 0` is an error.
## f2 — the four bulk-load methods
| method | emits | use for |
|---|---|---|
| `insert_vertices(label, &batches)` | `UNWIND … CREATE (n:`Label`)` | single-label nodes |
| `insert_vertices_multi(&[l1, l2, …], &batches)` | `CREATE (n:`A`:`B`:`C`)` | multi-label nodes (single-label delegates here) |
| `insert_edges(src, type, dst, &batches)` | `UNWIND … MATCH endpoints on `_gar_id` … CREATE` | typed edges keyed on `_gar_id` (batches carry `src`/`dst`) |
| `insert_edges_keyed(src, type, dst, &[k…], &[k…], &batches)` | `MATCH (a:`L`{k1:…,k2:…})` | edges on a composite key — N props per endpoint |
```rust
// Multi-label:
loader.insert_vertices_multi(&["person", "employee"], &batches).await?;
// Composite-key edges — batch carries columns `country` and `code` per endpoint:
loader.insert_edges_keyed(
"city", "IN", "country",
&["country", "code"], &["country", "code"],
&edge_batches,
).await?;
```
### Going multicore — the `_par` bulk methods
The serial methods above await one query at a time over the single connection, so a big
initial load sits at ~1 core. For fresh/bulk loads use the concurrent siblings — same
Cypher, same result, up to `concurrency` queries in flight over the (cloned) multiplexed
link:
```rust
let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4);
// &self (not &mut) — each in-flight task clones the connection.
loader.insert_vertices_par("person", &vertex_batches, cores).await?;
// vertices are a barrier before edges (edges MATCH endpoints that must exist):
loader.insert_edges_par("person", "KNOWS", "person", &edge_batches, cores).await?;
// progress-aware variants take an `on_progress(rows)` closure fired per window:
loader.insert_vertices_multi_par(&["person"], &vertex_batches, cores, |n| bump(n)).await?;
```
The produced graph is identical to the serial path (vertices `CREATE` over disjoint
`_gar_id` ranges; execution order is irrelevant). Keep the serial `insert_vertices` when
you need deterministic single-link ordering.
Every `_par` method dispatches through one generic scheduler, `buffered_dispatch(items,
concurrency, run_one, on_progress)` — up to `concurrency` futures in flight via
`buffer_unordered`. It is connection-free so the "really runs `concurrency` at once"
contract is proven without a live DB (`render_tests::buffered_dispatch_runs_windows_concurrently`
records the peak in-flight count and asserts it hits `concurrency`; a regression to a serial
loop drops the peak to 1 and fails). The `knut.falkor_parallel_dispatch` bench times the
serial loop vs this scheduler and reports the `speedup_x`.
Key columns may be any Cypher-renderable Arrow type, not only `Int64`. Any column not named
by a key is treated as an edge property.
## f2b — `GRAPH.BULK`, the fast path (≈48× the parallel `UNWIND`)
The `_par` methods above are **not** the fastest way to fill an empty graph. `GRAPH.BULK` is
a pure-Rust binary encoder that renders Arrow batches straight onto FalkorDB's bulk wire
protocol as **one RESP command** — no Cypher parsing, no CSV, no Python, no
`falkordb_bulk_loader` shell-out:
```rust
use graphar_falkordb::{FalkorDbLoader, BulkNodeGroup, BulkEdgeGroup, choose_bulk};
let mut loader = FalkorDbLoader::connect("redis://127.0.0.1:6379", "social").await?;
// Gate it yourself, or let choose_bulk decide:
if choose_bulk(loader.graph_is_empty().await) {
loader.bulk_insert_streamed(&node_groups, &edge_groups).await?;
} else {
// non-empty target: fall back to the UNWIND lanes above
}
```
| call | does |
|---|---|
| `bulk_insert(&nodes, &edges)` | encode + ship one `GRAPH.BULK` payload |
| `bulk_insert_streamed(&nodes, &edges)` | same, chunked per blob — prefer this |
| `choose_bulk(graph_is_empty)` | the selection predicate the real consumers use |
| `graph_is_empty()` / `label_is_empty(l)` | the emptiness probes that gate it |
Measured (`bench_bulk_vs_unwind`, 200k nodes, **debug** build, one FalkorDB container):
serial `UNWIND` ~25k nodes/s → parallel `UNWIND` ~56k → **`GRAPH.BULK` ~2.69M** (≈48× the
parallel path, ≈108× the baseline). Treat the absolute figures as lower bounds and the
ratios as the real result.
**Constraints:** it `CREATE`s, so it cannot upsert and errors on `BEGIN` against an existing
key — hence the empty-graph gate. It is not transactional. Wire format and the byte-exact
DB-free encoder tests: [`falkordb-bulk-design.md`](falkordb-bulk-design.md).
You usually do not call this yourself: both real consumers — `knut-popviz` and
`knut-bifrost`'s sink — auto-select it via `choose_bulk` and fall back to `UNWIND`
automatically.
## f3 — the population example
The `population` example is the whole path end to end — a Rust port of
`simulate_population.py` that generates the 12-label / 17-edge Swedish-population graph as
**Arrow** (typed + columnar, no CSV) and bulk-loads it:
```
cargo run -p graphar-falkordb --release --example population
```
Concurrency shape, for the record: the loader uses **one multiplexed connection, cloned per
in-flight future** (`buffered_dispatch` + `buffer_unordered`) — *not* a loader-per-connection
sharding scheme. Largest **measured** loads are 200k nodes (`bench_bulk_vs_unwind`) and
60,600 vertices (`bench_popviz_bulk_vs_unwind`); anything beyond that is an extrapolated
design target, not a recorded result.
## Surfaces (Rust API)
| Call | Does |
|---|---|
| `FalkorDbLoader::connect(url, graph)` | Open a multiplexed Redis connection to a graph |
| `loader.with_batch_size(n)` | Override the 512-row default batch size (`0` errors) |
| `loader.insert_vertices(label, &batches)` | `UNWIND … CREATE` vertices of one label |
| `loader.insert_vertices_multi(&[l1, l2, …], &batches)` | `UNWIND … CREATE (n:`A`:`B`)` multi-label nodes |
| `loader.insert_edges(src, type, dst, &batches)` | `UNWIND … CREATE` typed edges keyed on `_gar_id` |
| `loader.insert_edges_keyed(src, type, dst, &[k…], &[k…], &batches)` | Edges matched on a composite key (N props / endpoint) |
| `loader.bulk_insert_streamed(&nodes, &edges)` | **`GRAPH.BULK` — the ≈48× fast path, empty graph only** (§f2b) |
| `choose_bulk(loader.graph_is_empty().await)` | Pick bulk vs `UNWIND` the way the real consumers do |
| `loader.create_index(label, prop)` | Range index — build before edges so endpoint lookups don't scan |
| `loader.run_query(cypher)` | Run an arbitrary Cypher write |
Every `insert_*` / `bulk_insert*` call above **`CREATE`s and is not idempotent** — see the
warning at the top of this file.