From d0613ddb91151413e13956cde714089bba8a4cde Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Mon, 31 Aug 2026 15:43:01 -0400 Subject: [PATCH] fix(neo4j): stream the Cypher snapshot instead of building it as one String --emit neo4j died on a large repository before writing a byte: OutOfMemoryError: UTF16 String size is 1169246895, should be less than 1073741823 at CypherWriter.renderCypher(CypherWriter.java:82) Not heap exhaustion, so no -Xmx avoided it: renderCypher collected every statement into a list and collapsed it with String.join, and the result has to fit in one array. A script carrying any non-Latin-1 character is stored as UTF16, which halves the ceiling to ~1.07 billion characters. ThingsBoard v4.0 renders 1.169 billion, so the join threw and left a zero-byte graph.cypher. CypherWriter now streams: writeCypher(Appendable, rows, appName) emits each statement as it is produced, so nothing larger than one 500-row batch is ever a String and peak memory no longer scales with the graph. renderCypher stays as a thin wrapper over it for tests and small callers, documented as unsuitable for user-facing output, so the two cannot drift. Statements are separated by newlines rather than terminated by them, matching String.join exactly. Getting this wrong appends one byte, which a test comparing renderCypher against writeCypher cannot see now that the former delegates to the latter -- so the boundary is pinned directly, and the whole script was diffed against the v3.0.0 release build. Neo4jEmitter writes through Files.newBufferedWriter in explicit UTF-8. It used a bare FileWriter, which encodes in the platform default charset; on Java 11 that is not UTF-8 everywhere, so non-ASCII could already be mangled. Verified on ThingsBoard v4.0 (4131 files): the v3.0.0 jar exits 1 with the OOM above and writes 0 bytes; this build exits 0 in 5m20s and writes a 1.17 GB script whose 4063 UNWIND blocks are all balanced. On daytrader8 the emitted graph.cypher is byte-identical to the release. Closes #209 --- .../java/com/ibm/cldk/neo4j/CypherWriter.java | 94 +++++++++++---- .../java/com/ibm/cldk/neo4j/Neo4jEmitter.java | 13 +- .../cldk/neo4j/CypherWriterStreamingTest.java | 113 ++++++++++++++++++ 3 files changed, 194 insertions(+), 26 deletions(-) create mode 100644 src/test/java/com/ibm/cldk/neo4j/CypherWriterStreamingTest.java diff --git a/src/main/java/com/ibm/cldk/neo4j/CypherWriter.java b/src/main/java/com/ibm/cldk/neo4j/CypherWriter.java index 7a6116a7..bc34479b 100644 --- a/src/main/java/com/ibm/cldk/neo4j/CypherWriter.java +++ b/src/main/java/com/ibm/cldk/neo4j/CypherWriter.java @@ -14,6 +14,8 @@ import com.ibm.cldk.neo4j.GraphRows.EdgeRow; import com.ibm.cldk.neo4j.GraphRows.NodeRow; +import java.io.IOException; +import java.io.StringWriter; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; @@ -55,31 +57,81 @@ public final class CypherWriter { private CypherWriter() {} + /** + * Render the whole script into a {@code String}. Convenience for callers holding a small graph + * (tests, diagnostics); it delegates to {@link #writeCypher} so the two can never drift. + * + *

Prefer {@link #writeCypher} for anything user-facing. A large repository's script + * exceeds the JVM's maximum {@code String} length and this method then throws + * {@code OutOfMemoryError: Requested string length exceeds VM limit} — a ceiling on one array, + * not a heap shortage, so no {@code -Xmx} avoids it (#209). + */ public static String renderCypher(GraphRows rows, String appName) { - List out = new ArrayList<>(); + StringWriter out = new StringWriter(); + try { + writeCypher(out, rows, appName); + } catch (IOException impossible) { + // StringWriter never throws; it only declares IOException to satisfy Writer. + throw new IllegalStateException(impossible); + } + return out.toString(); + } - out.add("// ── constraints & indexes ──"); + /** + * Stream the script to {@code out}, one statement at a time. Nothing larger than a single batch + * is ever held as a {@code String}, so peak memory is bounded by {@link #BATCH} rather than by + * the size of the graph. + * + *

{@code out} is written to incrementally and is not flushed or closed here — the caller owns + * it, and should hand in a buffered writer so that per-statement writes do not become + * per-statement syscalls. + */ + public static void writeCypher(Appendable out, GraphRows rows, String appName) throws IOException { + Statements s = new Statements(out); + s.add("// ── constraints & indexes ──"); for (String stmt : Schema.CONSTRAINTS) { - out.add(stmt + ";"); + s.add(stmt + ";"); } for (String stmt : Schema.INDEXES) { - out.add(stmt + ";"); + s.add(stmt + ";"); } - out.add(""); - out.add("// ── wipe this project's prior subgraph (packages/annotations/artifacts/config keys are shared) ──"); - out.add(wipe(appName)); + s.add(""); + s.add("// ── wipe this project's prior subgraph (packages/annotations/artifacts/config keys are shared) ──"); + s.add(wipe(appName)); - out.add(""); - out.add("// ── nodes ──"); - out.addAll(nodeStatements(rows.nodes)); + s.add(""); + s.add("// ── nodes ──"); + writeNodeStatements(s, rows.nodes); - out.add(""); - out.add("// ── relationships ──"); - out.addAll(edgeStatements(rows.edges)); + s.add(""); + s.add("// ── relationships ──"); + writeEdgeStatements(s, rows.edges); + + s.add(""); + } - out.add(""); - return String.join("\n", out); + /** + * Writes statements separated by newlines, exactly as {@code String.join("\n", ...)} did: + * the separator goes BETWEEN statements, so the script does not gain a trailing newline the + * rendered form never had. Emitting {@code statement + "\n"} each time would append one extra + * byte at the end — small, but this file is compared byte-for-byte across versions. + */ + private static final class Statements { + private final Appendable out; + private boolean first = true; + + Statements(Appendable out) { + this.out = out; + } + + void add(String statement) throws IOException { + if (!first) { + out.append('\n'); + } + first = false; + out.append(statement); + } } private static String wipe(String appName) { @@ -103,14 +155,13 @@ private static String wipe(String appName) { // Nodes — grouped by their full label set + key property, batched into UNWIND lists. // ---------------------------------------------------------------------------------------------- - private static List nodeStatements(List nodes) { + private static void writeNodeStatements(Statements out, List nodes) throws IOException { Map> groups = new LinkedHashMap<>(); for (NodeRow n : nodes) { String k = String.join(":", n.labels) + "|" + n.keyProp; groups.computeIfAbsent(k, x -> new ArrayList<>()).add(n); } - List blocks = new ArrayList<>(); for (List group : groups.values()) { NodeRow head = group.get(0); String mergeLabel = head.labels.get(0); @@ -121,19 +172,18 @@ private static List nodeStatements(List nodes) { for (NodeRow n : batch) { list.add(" {k: " + cypherValue(n.value) + ", p: " + cypherMap(n.props) + "}"); } - blocks.add("UNWIND [\n" + String.join(",\n", list) + "\n] AS row\n" + out.add("UNWIND [\n" + String.join(",\n", list) + "\n] AS row\n" + "MERGE (n:" + mergeLabel + " {" + head.keyProp + ": row.k})\n" + "SET n += row.p" + setLabels + ";"); } } - return blocks; } // ---------------------------------------------------------------------------------------------- // Edges — grouped by (type, endpoint labels + key props), batched. // ---------------------------------------------------------------------------------------------- - private static List edgeStatements(List edges) { + private static void writeEdgeStatements(Statements out, List edges) throws IOException { Map> groups = new LinkedHashMap<>(); for (EdgeRow e : edges) { String k = e.type + "|" + e.from.label + "." + e.from.keyProp + "|" + e.to.label + "." + e.to.keyProp @@ -141,7 +191,6 @@ private static List edgeStatements(List edges) { groups.computeIfAbsent(k, x -> new ArrayList<>()).add(e); } - List blocks = new ArrayList<>(); for (List group : groups.values()) { EdgeRow head = group.get(0); boolean keyed = head.key != null; @@ -152,14 +201,13 @@ private static List edgeStatements(List edges) { + (keyed ? ", k: " + cypherValue(e.key) : "") + ", p: " + cypherMap(e.props) + "}"); } - blocks.add("UNWIND [\n" + String.join(",\n", list) + "\n] AS row\n" + out.add("UNWIND [\n" + String.join(",\n", list) + "\n] AS row\n" + "MATCH (a:" + head.from.label + " {" + head.from.keyProp + ": row.f})\n" + "MATCH (b:" + head.to.label + " {" + head.to.keyProp + ": row.t})\n" + "MERGE (a)-[r:" + head.type + (keyed ? " {_k: row.k}" : "") + "]->(b)\n" + "SET r += row.p;"); } } - return blocks; } // ---------------------------------------------------------------------------------------------- diff --git a/src/main/java/com/ibm/cldk/neo4j/Neo4jEmitter.java b/src/main/java/com/ibm/cldk/neo4j/Neo4jEmitter.java index 8264681b..e68f5df9 100644 --- a/src/main/java/com/ibm/cldk/neo4j/Neo4jEmitter.java +++ b/src/main/java/com/ibm/cldk/neo4j/Neo4jEmitter.java @@ -19,6 +19,8 @@ import java.io.File; import java.io.FileWriter; import java.io.IOException; +import java.io.Writer; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; @@ -100,10 +102,15 @@ private static void write(GraphRows rows, String name, String output, boolean ta String dir = output != null ? output : System.getProperty("user.dir"); Files.createDirectories(Paths.get(dir)); - try (FileWriter w = new FileWriter(new File(dir, "graph.cypher"))) { - w.write(CypherWriter.renderCypher(rows, name)); - Log.done("Neo4j graph.cypher saved at " + dir + File.separator + "graph.cypher"); + // Streamed, not rendered: a large repository's script exceeds the JVM's maximum String + // length, and building it whole threw OutOfMemoryError before a byte was written (#209). + // UTF-8 explicitly rather than FileWriter's platform default, so non-ASCII in identifiers + // or captured source survives on a machine whose default encoding is not UTF-8. + Path cypher = Paths.get(dir).resolve("graph.cypher"); + try (Writer w = Files.newBufferedWriter(cypher, StandardCharsets.UTF_8)) { + CypherWriter.writeCypher(w, rows, name); } + Log.done("Neo4j graph.cypher saved at " + cypher); } /** diff --git a/src/test/java/com/ibm/cldk/neo4j/CypherWriterStreamingTest.java b/src/test/java/com/ibm/cldk/neo4j/CypherWriterStreamingTest.java new file mode 100644 index 00000000..7114ca85 --- /dev/null +++ b/src/test/java/com/ibm/cldk/neo4j/CypherWriterStreamingTest.java @@ -0,0 +1,113 @@ +package com.ibm.cldk.neo4j; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.ibm.cldk.neo4j.GraphRows.NodeRef; +import java.io.IOException; +import java.io.StringWriter; +import java.io.Writer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +/** + * The snapshot writer must stream. Building the whole script as one {@code String} first is what + * made {@code --emit neo4j} die on a large repository with {@code OutOfMemoryError: Requested + * string length exceeds VM limit} (#209) — not heap exhaustion, but a single {@code String} + * exceeding the JVM's maximum array length, which no {@code -Xmx} can raise. + */ +class CypherWriterStreamingTest { + + /** A graph big enough to span several batches, so batching and streaming both get exercised. */ + private static GraphRows sampleRows(int nodes) { + RowBuilder b = new RowBuilder(); + List refs = new ArrayList<>(); + for (int i = 0; i < nodes; i++) { + Map p = RowBuilder.props(); + p.put("name", "Type" + i); + p.put("code", "class Type" + i + " { void m() {} }"); + refs.add(b.node(Arrays.asList("JType", "JSymbol"), "id", "can://java/app/T" + i + ".java/Type" + i, p)); + } + for (int i = 1; i < nodes; i++) { + b.edge("J_CALLS", refs.get(i - 1), refs.get(i)); + } + return b.finish(); + } + + /** Records every write so the test can tell streaming from one giant write. */ + private static final class CountingWriter extends Writer { + final StringWriter sink = new StringWriter(); + int writes; + int largestWrite; + + @Override + public void write(char[] cbuf, int off, int len) throws IOException { + writes++; + largestWrite = Math.max(largestWrite, len); + sink.write(cbuf, off, len); + } + + @Override + public void flush() {} + + @Override + public void close() {} + } + + @Test + void streamedOutputIsIdenticalToTheRenderedString() throws IOException { + GraphRows rows = sampleRows(1200); // > 2 batches of 500 + StringWriter w = new StringWriter(); + CypherWriter.writeCypher(w, rows, "app"); + assertEquals(CypherWriter.renderCypher(rows, "app"), w.toString(), + "streaming must not change a single byte of the emitted script"); + } + + @Test + void theScriptReachesTheWriterIncrementallyRatherThanAsOneString() throws IOException { + GraphRows rows = sampleRows(1200); + CountingWriter w = new CountingWriter(); + CypherWriter.writeCypher(w, rows, "app"); + + int total = w.sink.toString().length(); + assertTrue(total > 100_000, "sample graph should be substantial, was " + total + " chars"); + // The defect being pinned: render-then-write hands the writer everything in one call. Any + // streaming implementation writes each statement separately, so no single write is anywhere + // near the whole script. + assertTrue(w.writes > 10, "expected many writes, got " + w.writes); + assertTrue(w.largestWrite < total / 2, + "no single write may carry half the script (largest=" + w.largestWrite + " of " + total + + ") — that means the whole script was materialized before writing"); + } + + /** + * The equality test above cannot catch a separator regression, because {@code renderCypher} now + * delegates to {@code writeCypher} and so compares the implementation with itself. This pins the + * boundary independently: statements are separated by newlines, not terminated by them. + * Emitting {@code statement + "\n"} per statement appends one byte the previous released + * version never wrote, which is invisible to a self-comparison and obvious in a diff of two + * generated scripts. + */ + @Test + void statementsAreSeparatedByNewlinesNotTerminatedByThem() throws IOException { + StringWriter w = new StringWriter(); + CypherWriter.writeCypher(w, sampleRows(3), "app"); + String out = w.toString(); + assertTrue(out.endsWith(";\n"), "script ends with the final statement then one newline"); + assertFalse(out.endsWith("\n\n"), "no trailing blank line — the separator goes between statements"); + } + + @Test + void anEmptyGraphStillEmitsConstraintsAndTheWipe() throws IOException { + StringWriter w = new StringWriter(); + CypherWriter.writeCypher(w, new RowBuilder().finish(), "app"); + String out = w.toString(); + assertTrue(out.contains("CREATE CONSTRAINT"), "constraints are emitted even with no rows"); + assertTrue(out.contains("DETACH DELETE"), "the wipe is emitted even with no rows"); + assertEquals(CypherWriter.renderCypher(new RowBuilder().finish(), "app"), out); + } +}