nordisk/knut

Arrow → FalkorDB bulk load — guide

Class: guide · audience: a techie bulk-loading Arrow batches into FalkorDB. The what is in falkordb-design.md; the why in falkordb-reasoning.md; the fastest path, GRAPH.BULK, is in falkordb-bulk-design.md. This file is the API and the load workflow.

⚠ Nothing in this crate is idempotent. Every method below CREATEs — 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 §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

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

// 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:

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:

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
}

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 CREATEs, 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.

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)

Every insert_* / bulk_insert* call above CREATEs and is not idempotent — see the warning at the top of this file.