nordisk/knut

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). Strict: this file states only the shape we run. The why is in falkordb-reasoning.md; the API in 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:

Reuse map: it uses graphar 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:

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:

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.