From 906c720676b2c3231fcda9397ca0e10beb320643 Mon Sep 17 00:00:00 2001 From: Bonobo Date: Wed, 2 Sep 2026 13:20:14 +0200 Subject: [PATCH] perf(neo4j): stream Cypher snapshots to disk --- src/build/neo4j/cypher.ts | 77 ++++++++++++++++++++++++--------------- src/build/neo4j/index.ts | 2 +- src/utils/serialize.ts | 4 +- test/neo4j-schema.test.ts | 16 ++++++++ 4 files changed, 67 insertions(+), 32 deletions(-) diff --git a/src/build/neo4j/cypher.ts b/src/build/neo4j/cypher.ts index 986d7b0..72efee7 100644 --- a/src/build/neo4j/cypher.ts +++ b/src/build/neo4j/cypher.ts @@ -7,30 +7,47 @@ * it expresses the full truth. Incremental updates are the bolt writer's job. */ +import * as fs from "node:fs"; import type { EdgeRow, GraphRows, NodeRow, Props } from "./rows"; -import { chunk, cypherMap, cypherValue } from "./rows"; +import { cypherMap, cypherValue } from "./rows"; import { CONSTRAINTS, INDEXES } from "./schema"; const BATCH = 500; export function renderCypher(rows: GraphRows, appId: string): string { - const out: string[] = []; + return [...cypherBlocks(rows, appId)].join("\n"); +} - out.push("// ── constraints & indexes ──"); - for (const stmt of CONSTRAINTS) out.push(`${stmt};`); - for (const stmt of INDEXES) out.push(`${stmt};`); +export function writeCypherFile(filePath: string, rows: GraphRows, appId: string): void { + const file = fs.openSync(filePath, "w"); + try { + let first = true; + for (const block of cypherBlocks(rows, appId)) { + fs.writeFileSync(file, first ? block : `\n${block}`); + first = false; + } + } finally { + fs.closeSync(file); + } +} - out.push("", "// ── wipe this project's prior subgraph (external targets are shared) ──"); - out.push(wipe(appId)); +function* cypherBlocks(rows: GraphRows, appId: string): Generator { + yield "// ── constraints & indexes ──"; + for (const stmt of CONSTRAINTS) yield `${stmt};`; + for (const stmt of INDEXES) yield `${stmt};`; - out.push("", "// ── nodes ──"); - for (const block of nodeStatements(rows.nodes)) out.push(block); + yield ""; + yield "// ── wipe this project's prior subgraph (external targets are shared) ──"; + yield wipe(appId); - out.push("", "// ── relationships ──"); - for (const block of edgeStatements(rows.edges)) out.push(block); + yield ""; + yield "// ── nodes ──"; + yield* nodeStatements(rows.nodes); - out.push(""); - return out.join("\n"); + yield ""; + yield "// ── relationships ──"; + yield* edgeStatements(rows.edges); + yield ""; } function wipe(appId: string): string { @@ -47,63 +64,65 @@ function wipe(appId: string): string { // Nodes — grouped by their full label set + key property, batched into UNWIND lists. // ---------------------------------------------------------------------------------------------- -function nodeStatements(nodes: NodeRow[]): string[] { +function* nodeStatements(nodes: NodeRow[]): Generator { const groups = new Map(); for (const n of nodes) { const k = `${n.labels.join(":")}|${n.keyProp}`; (groups.get(k) ?? groups.set(k, []).get(k)!).push(n); } - const blocks: string[] = []; for (const group of groups.values()) { const { labels, keyProp } = group[0]; const mergeLabel = labels[0]; const extra = labels.slice(1); const setLabels = extra.length ? `, n:${extra.join(":")}` : ""; - for (const batch of chunk(group, BATCH)) { + for (const batch of batches(group, BATCH)) { const list = batch .map((n) => ` {k: ${cypherValue(n.value)}, p: ${cypherMap(n.props)}}`) .join(",\n"); - blocks.push( + yield ( `UNWIND [\n${list}\n] AS row\n` + - `MERGE (n:${mergeLabel} {${keyProp}: row.k})\n` + - `SET n += row.p${setLabels};`, + `MERGE (n:${mergeLabel} {${keyProp}: row.k})\n` + + `SET n += row.p${setLabels};` ); } } - return blocks; } // ---------------------------------------------------------------------------------------------- // Edges — grouped by (type, endpoint labels + key props), batched. // ---------------------------------------------------------------------------------------------- -function edgeStatements(edges: EdgeRow[]): string[] { +function* edgeStatements(edges: EdgeRow[]): Generator { const groups = new Map(); for (const e of edges) { const k = `${e.type}|${e.from.label}.${e.from.keyProp}|${e.to.label}.${e.to.keyProp}|${e.key !== undefined}`; (groups.get(k) ?? groups.set(k, []).get(k)!).push(e); } - const blocks: string[] = []; for (const group of groups.values()) { const { type, from, to } = group[0]; // Discriminated relationships MERGE on `{_k}` — see EdgeRow.key (issue #70). const keyed = group[0].key !== undefined; - for (const batch of chunk(group, BATCH)) { + for (const batch of batches(group, BATCH)) { const list = batch .map((e) => ` {f: ${cypherValue(e.from.value)}, t: ${cypherValue(e.to.value)}, ${keyed ? `k: ${cypherValue(e.key!)}, ` : ""}p: ${cypherMap(e.props)}}`) .join(",\n"); - blocks.push( + yield ( `UNWIND [\n${list}\n] AS row\n` + - `MATCH (a:${from.label} {${from.keyProp}: row.f})\n` + - `MATCH (b:${to.label} {${to.keyProp}: row.t})\n` + - `MERGE (a)-[r:${type}${keyed ? " {_k: row.k}" : ""}]->(b)\n` + - `SET r += row.p;`, + `MATCH (a:${from.label} {${from.keyProp}: row.f})\n` + + `MATCH (b:${to.label} {${to.keyProp}: row.t})\n` + + `MERGE (a)-[r:${type}${keyed ? " {_k: row.k}" : ""}]->(b)\n` + + `SET r += row.p;` ); } } - return blocks; +} + +function* batches(items: T[], size: number): Generator { + for (let offset = 0; offset < items.length; offset += size) { + yield items.slice(offset, offset + size); + } } // Re-exported for the bolt writer (which batches the same rows but binds them as params). diff --git a/src/build/neo4j/index.ts b/src/build/neo4j/index.ts index e94820d..fb24834 100644 --- a/src/build/neo4j/index.ts +++ b/src/build/neo4j/index.ts @@ -1,7 +1,7 @@ // Neo4j output: pure projection of the TSApplication IR to graph rows, plus the two writers // (cypher snapshot / bolt incremental). Nothing here runs unless `--emit neo4j` is selected. export { project } from "./project"; -export { renderCypher } from "./cypher"; +export { renderCypher, writeCypherFile } from "./cypher"; export { boltWriter, type BoltConfig } from "./bolt"; export { SCHEMA_VERSION, TS_PREFIX, buildSchemaDocument, NODE_LABELS, REL_TYPES, MARKER_LABELS, CONSTRAINTS, INDEXES } from "./schema"; export type { SchemaDocument } from "./schema"; diff --git a/src/utils/serialize.ts b/src/utils/serialize.ts index 95b9989..09ce442 100644 --- a/src/utils/serialize.ts +++ b/src/utils/serialize.ts @@ -1,6 +1,6 @@ import * as fs from "node:fs"; import * as path from "node:path"; -import { boltWriter, buildSchemaDocument, project, renderCypher } from "../build/neo4j"; +import { boltWriter, buildSchemaDocument, project, writeCypherFile } from "../build/neo4j"; import type { AnalysisOptions } from "../options"; import type { TSAnalysis } from "../schema"; import { Logger } from "./logging"; @@ -64,5 +64,5 @@ async function emitNeo4j(application: TSAnalysis, opts: AnalysisOptions): Promis const dir = opts.output ?? process.cwd(); fs.mkdirSync(dir, { recursive: true }); - fs.writeFileSync(path.join(dir, "graph.cypher"), renderCypher(rows, appId)); + writeCypherFile(path.join(dir, "graph.cypher"), rows, appId); } diff --git a/test/neo4j-schema.test.ts b/test/neo4j-schema.test.ts index 8872c94..b261f09 100644 --- a/test/neo4j-schema.test.ts +++ b/test/neo4j-schema.test.ts @@ -16,6 +16,8 @@ import { REL_TYPES, buildSchemaDocument, project, + renderCypher, + writeCypherFile, } from "../src/build/neo4j"; import { analyze } from "../src/core"; import type { AnalysisOptions } from "../src/options"; @@ -188,3 +190,17 @@ describe("neo4j inheritance edges (issue #33)", () => { expect(nodeValues.has(shape!.value)).toBe(true); }); }); + +test("streamed Cypher snapshot matches the compatibility renderer byte for byte", () => { + const application = rows.nodes.find((node) => node.labels.includes("TSApplication")); + expect(application, "TSApplication row").toBeDefined(); + + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "cants-cypher-test-")); + const output = path.join(dir, "graph.cypher"); + try { + writeCypherFile(output, rows, application!.value); + expect(fs.readFileSync(output, "utf8")).toBe(renderCypher(rows, application!.value)); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } +});