# Arrow → FalkorDB bulk load — design
> **Class:** `design` · **Function axis:** `f0` (the load bridge) + `f1` loader · `f2` Cypher rendering · `f3` UNWIND batching · `f4` edges on `_gar_id` · `f5` multi-label + composite key.
> Reuse graph: `falkordb.uses = [graphar]` — endpoint identity is graphar's `_gar_id` (see [`graphar-design.md`](graphar-design.md)).
> Strict: this file states *only the shape we run*. The *why* is in [`falkordb-reasoning.md`](falkordb-reasoning.md); the API in [`falkordb-guide.md`](falkordb-guide.md).
## f0 — the load bridge (the whole)
`graphar-falkordb` is the write bridge from the columnar world to FalkorDB
(`crates/graphar-falkordb/src`). It renders Arrow `RecordBatch` rows as Cypher map literals
and ships each batch with one `UNWIND [ {..}, {..} ] AS r CREATE/MERGE …`, so a whole chunk
becomes nodes or relationships in a single round trip. FalkorDB is reached as a Redis module
(`GRAPH.QUERY` over RESP), so the only dependency is an async Redis client.
Three laws hold across the bridge:
| law | statement | enforced by |
|---|---|---|
| **F-1** | One batch = one `UNWIND … AS r` round trip | `unwind_create_query` / `unwind_edge_query` (`src/cypher.rs:320`,`:353`) |
| **F-2** | Arrow → Cypher is explicit and **total** — unmapped type is a loud error | `arrow_value_to_cypher` errors with `FalkorError::UnsupportedType` (`src/cypher.rs:137`, `src/error.rs:20`) |
| **F-3** | The hot path is column-resolved — downcast once per column, no per-cell `String` | `ColRenderer::resolve` (`src/cypher.rs:31`) + `write_row_map` (`:98`) |
**Reuse map:** it `uses` [`graphar`](graphar-design.md) for the `_gar_id` endpoint identity
and `arrow` (58) for the batches. It is used *by* `graphar-flight` (Flight SQL DoPut loads
through this same loader) and `knut-bifrost` (`FalkorDbSink`).
## f1 — the loader
`FalkorDbLoader` (`src/loader.rs:21`) owns a multiplexed Redis connection to one graph:
| fn | signature | does |
|---|---|---|
| `connect` | `connect(url, graph) -> Result<Self>` (`src/loader.rs:28`) | open a multiplexed async Redis connection |
| `with_batch_size` | `with_batch_size(n) -> Result<Self>` (`:38`) | override `DEFAULT_BATCH_SIZE` (`:14`, = 512); `0` errors |
| `insert_vertices` | `insert_vertices(label, &batches) -> usize` (`:53`) | `UNWIND … CREATE` single-label nodes; delegates to `_multi` |
| `insert_vertices_multi` | `insert_vertices_multi(&[labels], &batches)` (`:64`) | `UNWIND … CREATE (n:`A`:`B`)` multi-label nodes |
| `insert_edges` | `insert_edges(src, type, dst, &batches)` (`:139`) | typed edges; delegates to `_keyed` with `["_gar_id"]` |
| `insert_edges_keyed` | `insert_edges_keyed(src, type, dst, &[k…], &[k…], &batches)` (`:169`) | edges matched on a composite key |
| `insert_vertices_par` | `insert_vertices_par(&self, label, &batches, concurrency) -> usize` | **concurrent** sibling of `insert_vertices` — up to `concurrency` in-flight |
| `insert_vertices_multi_par` | `insert_vertices_multi_par(&self, &[labels], &batches, concurrency, on_progress)` | concurrent multi-label + per-window `on_progress(rows)` |
| `insert_edges_par` / `insert_edges_keyed_par` | `(…, concurrency, on_progress)` | concurrent siblings of the edge inserts |
| `buffered_dispatch` | `buffered_dispatch(items, concurrency, run_one, on_progress) -> Result<()>` (free fn) | the generic, connection-free scheduler every `_par` method routes through — up to `concurrency` futures in flight (`buffer_unordered`); unit-tested that it truly overlaps |
| `create_index` | `create_index(label, prop)` (`:317`) | `CREATE INDEX FOR (n:`L`) ON (n.`prop`)` |
| `run_query` | `run_query(cypher) -> redis::Value` (`:307`) | arbitrary Cypher write |
Each inserted node gets a `_gar_id` property = its absolute 0-based row index across all
batches, matching graphar's internal vertex id.
### f3a — the parallel bulk path (multicore ingest)
The serial `insert_*` methods `await` one `UNWIND … CREATE` query at a time over the
single `MultiplexedConnection`, so the client idles a full round-trip between queries —
the ingest is pinned at **~1 core** regardless of box size. `MultiplexedConnection` is
`Clone` and multiplexes many concurrent in-flight commands over the one link, so the
`_par` methods (`&self`) split the work identically to the serial path — the **same**
per-window Cypher, via the shared `render_vertex_queries` / `render_edge_queries` +
`build_vertex_query` renderers (byte-identical, same `_gar_id` scheme) — and dispatch it
through the generic `buffered_dispatch` (which wraps `futures::stream::buffer_unordered(concurrency)`)
over cloned connections (`dispatch_par` supplies only the per-window work — clone the conn,
run one `GRAPH.QUERY`), saturating the pipeline. Because vertices `CREATE` over disjoint
`_gar_id` ranges and edges `MATCH`-then-`CREATE` existing endpoints, execution order is
irrelevant: the **produced graph is identical** to the serial path (proven by
`render_tests` exact-string/count asserts, and live by
`integration_falkordb::parallel_bulk_matches_serial_counts`). Additive — the serial
methods are untouched; callers/tests that want deterministic single-link ordering keep
using them. Emits `insert_vertices_par` / `insert_edges_par` matrix cells.
**Concurrency proven DB-free.** `buffered_dispatch` is generic over the per-item work
precisely so the core-saturation contract is testable without a live FalkorDB:
`render_tests::buffered_dispatch_runs_windows_concurrently` feeds a fake `run_one` that
records the peak simultaneous in-flight count and asserts it reaches `concurrency` (24
windows @ 8 → peak 8). This is the **red-when-broken** guard for the fix: a regression to a
serial `for … await` collapses the peak to 1 and fails the test (verified). A current-thread
runtime is used on purpose — it is exactly the single-reactor / many-in-flight-commands model
of `MultiplexedConnection`, so the overlap proven is the overlap that saturates the real link.
The **light DB-free bench** `knut.falkor_parallel_dispatch` (xtask suite) times the pre-fix
serial one-RTT-at-a-time loop against `buffered_dispatch` at `available_parallelism()` and
reports `speedup_x` (→ ~concurrency); the authoritative live nodes/s + `cores_busy` on a real
container is the Loki bencher's / the `#[ignore]` integration path's.
### f3b — index BEFORE edges (the server-side half of the ingest speedup)
Client-side pipelining (f3a) removes the *client's* one-RTT-at-a-time stall, but it cannot
touch a stall the *server* creates. Every edge query `MATCH (a:`L` {`_gar_id`: …}),
(b:`L` {`_gar_id`: …}) CREATE …` resolves both endpoints by `_gar_id`. With **no index** on
`_gar_id`, each MATCH is a full **O(V) label scan**, so the whole edge phase is
**O(edges × vertices)** work on the server — a quadratic wall no amount of client concurrency
can climb (throwing more in-flight queries at a server doing per-edge full scans just queues
them). The fix is ordering: the popviz load driver (`knut_popviz::load_falkor`) builds the
`_gar_id` index on **every label BEFORE the edge phase**, so each endpoint MATCH is an indexed
**O(1)** lookup and the edge phase is O(edges). The canonical order —
`knut_popviz::falkor_load_phase_order()` = `[Vertices, Indexes, Edges]` — is the **single
source of truth** the driver iterates, so the sequence can't silently drift. Vertices come
first (nothing to index until the nodes exist); building the index does not slow the edge
`CREATE` (edges are relationships, `_gar_id` is a node property), so indexing early is a pure
win with no trade-off. **Red-when-broken (DB-free):**
`knut_popviz::tests::falkor_indexes_are_built_before_edges` asserts `Indexes` precedes `Edges`
in the order array — revert to indexes-after-edges and it fails. The graph produced is
byte-identical either way (index DDL never changes node/edge content); only wall-clock moves.
## f2 — Cypher rendering
`src/cypher.rs` turns Arrow into Cypher text. `arrow_value_to_cypher(array, row)`
(`src/cypher.rs:137`) is the **total** value map:
| Arrow type | → Cypher |
|---|---|
| `Boolean` | `true`/`false` |
| `Int32`/`Int64` | integer literal |
| `Float32`/`Float64` | float literal |
| `Utf8`/`LargeUtf8` | quoted string |
| `Date32`/`Date64`, `Timestamp` | ISO-8601 string |
| `List`/`LargeList` | Cypher list literal |
| anything else | `Err(FalkorError::UnsupportedType)` — never silent stringify |
`ColRenderer` (`src/cypher.rs:19`) is the column-resolved fast path: `resolve(array)` (`:31`)
downcasts each column once to its concrete Arrow array; `write(buf, row)` (`:56`) renders a
cell straight into a caller buffer. `write_row_map(buf, cols, row, extra)` (`:98`) renders a
whole `{ k: v, … }` map into one pre-sized buffer (the `extra` slot injects `_gar_id`);
`row_to_cypher_map` (`:127`) is the per-row convenience over it. Label rendering:
`render_labels(&[labels])` (`:329`) → `` `A`:`B`:`C` `` (a single label is plain `` `Label` ``).
## f3 — UNWIND batching
The query builders assemble one statement per batch of rendered maps:
- `unwind_create_query(label, &maps)` (`src/cypher.rs:320`) → `UNWIND $rows AS r CREATE (n:`L`) SET n = r`.
- `unwind_create_query_multi(&[labels], &maps)` (`:345`) → same with multi-label.
- `unwind_edge_query(...)` (`:353`) and `unwind_edge_query_keyed(...)` (`:408`) → `UNWIND … AS r MATCH (a:`L`{…}),(b:`L`{…}) CREATE (a)-[:`T`]->(b)`.
`DEFAULT_BATCH_SIZE` (`src/loader.rs:14`, = 512) rows amortise into one round trip; the loader
slices input batches into batch-size windows and issues one query per window.
## f4 — edges keyed on `_gar_id`
`insert_edges(src, type, dst, &batches)` (`src/loader.rs:139`) matches both endpoints by
`_gar_id`. It is a thin delegate to `insert_edges_keyed(src, type, dst, ["_gar_id"],
["_gar_id"], batches)` (`:169`) — for the single legacy `_gar_id` key the batch's carrier
columns are literally named `src`/`dst` (`EdgeRow`, `src/cypher.rs:459`). Endpoints are
resolved by MERGE/MATCH on `_gar_id`, so a range index on `_gar_id` keeps the lookup off a
full scan.
## f5 — multi-label & composite key
`insert_vertices_multi(&[labels], &batches)` (`src/loader.rs:64`) emits `CREATE (n:`A`:`B`)`
for nodes carrying several labels; `insert_vertices` is exactly the one-label slice and is
byte-identical to the historical single-label path (`render_labels` of one label is plain
`` `Label` ``, `src/cypher.rs:329`).
`insert_edges_keyed(src, type, dst, src_keys, dst_keys, &batches)` (`src/loader.rs:169`)
matches endpoints on a **composite key** — N properties per endpoint, e.g.
`MATCH (a:`L` {`k1`: …, `k2`: …})`. The carrier columns generalize from `src`/`dst` to one
column per named key (any Cypher-renderable Arrow type, not only `Int64`); a column not named
by a key is an edge property. The single-key `["_gar_id"]`/`["_gar_id"]` case is byte-identical
to `insert_edges`, still staging the `src`/`dst` carriers.