This is an automated email from the ASF dual-hosted git repository. spmallette pushed a commit to branch tinkergraph-storage in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit c2cf835ee829f73d43eae7792b4615f8f9abdf5f Author: Stephen Mallette <[email protected]> AuthorDate: Tue Aug 4 16:28:49 2026 +0000 Stream TinkerStorageGraph snapshots one element at a time Compaction built the entire graph snapshot as a single in-heap byte array (with a doubling buffer), so compacting a large graph needed a second full copy of it in memory and risked OOM. The snapshot is now streamed one element per framed record straight to the file, bounding peak memory to a single element. The on-disk format is unchanged: each frame is an ordinary single-entry put record that replay folds exactly as before. Write amplification (a commit rewrites each changed element in full) is documented as a known limitation rather than mitigated, since elements are small and auto-compaction bounds log growth. Assisted-by: Claude Code:claude-opus-4-8 --- .../structure/storage/GraphBinaryStorage.java | 51 +++++++++++----------- .../structure/storage/GraphBinaryStorageTest.java | 51 ++++++++++++++++++++++ 2 files changed, 77 insertions(+), 25 deletions(-) diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java index c9eecba9e5..074b67bc34 100644 --- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java +++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorage.java @@ -49,11 +49,9 @@ import java.nio.file.AtomicMoveNotSupportedException; import java.nio.file.Files; import java.nio.file.StandardCopyOption; import java.nio.file.StandardOpenOption; -import java.util.ArrayList; import java.util.Collection; import java.util.Iterator; import java.util.LinkedHashMap; -import java.util.List; import java.util.Map; /** @@ -68,6 +66,12 @@ import java.util.Map; * the operating system. * <p/> * The in-memory graph remains authoritative (write-through). This engine does not support graphs larger than memory. + * <p/> + * Known limitation (write amplification): a commit records each changed element in full — a single property change on + * a large element rewrites the whole element to the log. Elements are typically small and automatic compaction bounds + * the resulting log growth, so this is accepted rather than mitigated with per-property deltas, which would complicate + * the {@link TinkerStorageMutation} contract and the replay fold. The snapshot, by contrast, is streamed one element + * at a time so compaction never holds a second full copy of the graph in heap. */ public final class GraphBinaryStorage implements TinkerStorage { @@ -292,9 +296,7 @@ public final class GraphBinaryStorage implements TinkerStorage { final File tmp = new File(directory, SNAPSHOT_FILE + ".tmp"); try (final FileOutputStream fos = new FileOutputStream(tmp); final DataOutputStream out = new DataOutputStream(new BufferedOutputStream(fos))) { - final byte[] frame = encodeSnapshot(graph); - if (frame.length > 0) - writeFrame(out, frame); + writeSnapshot(graph, out); out.flush(); // force the snapshot's bytes to the device before it is renamed into place fos.getFD().sync(); @@ -357,34 +359,33 @@ public final class GraphBinaryStorage implements TinkerStorage { } /** - * Serialize the entire current committed state of the graph as a single put-only record. + * Write the entire current committed state of the graph to {@code out} as a stream of single-element put records, + * one frame per vertex and per edge. Each frame is an ordinary put record (see {@link #encodeRecord}) with an + * entry count of one, so {@link #foldRecords} reconstructs the graph from these frames exactly as it would from a + * commit log. Writing one element at a time keeps peak memory bounded to a single element rather than materializing + * the whole graph as one byte array, so a snapshot never needs to hold a second full copy of the graph in heap. */ - private byte[] encodeSnapshot(final AbstractTinkerGraph graph) throws IOException { - final List<Vertex> vertexList = new ArrayList<>(); + private void writeSnapshot(final AbstractTinkerGraph graph, final DataOutputStream out) throws IOException { final Iterator<Vertex> vertexIterator = graph.vertices(); while (vertexIterator.hasNext()) - vertexList.add(vertexIterator.next()); - final List<Edge> edgeList = new ArrayList<>(); + writeElementFrame(out, OP_PUT_VERTEX, vertexIterator.next()); final Iterator<Edge> edgeIterator = graph.edges(); while (edgeIterator.hasNext()) - edgeList.add(edgeIterator.next()); - - if (vertexList.isEmpty() && edgeList.isEmpty()) - return new byte[0]; + writeElementFrame(out, OP_PUT_EDGE, edgeIterator.next()); + } + /** + * Encode a single element as a one-entry put record and write it as a framed record to {@code out}. Only one + * element's bytes are held in memory at a time. + */ + private void writeElementFrame(final DataOutputStream out, final byte op, final Object element) throws IOException { final ByteBufferBuffer buffer = new ByteBufferBuffer(); buffer.writeByte(FORMAT_VERSION); - buffer.writeLong(0L); // snapshot has no single tx version - buffer.writeInt(vertexList.size() + edgeList.size()); - for (final Vertex v : vertexList) { - buffer.writeByte(OP_PUT_VERTEX); - writer.write(DetachedFactory.detach(v, true), buffer); - } - for (final Edge e : edgeList) { - buffer.writeByte(OP_PUT_EDGE); - writer.write(DetachedFactory.detach(e, true), buffer); - } - return buffer.toWrittenArray(); + buffer.writeLong(0L); // snapshot records have no single tx version + buffer.writeInt(1); + buffer.writeByte(op); + writer.write(DetachedFactory.detach(element, true), buffer); + writeFrame(out, buffer.toWrittenArray()); } @Override diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java index 8113e03d18..2d6f9952d5 100644 --- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/storage/GraphBinaryStorageTest.java @@ -20,10 +20,12 @@ package org.apache.tinkerpop.gremlin.tinkergraph.structure.storage; import org.apache.commons.configuration2.Configuration; import org.apache.tinkerpop.gremlin.structure.T; +import org.apache.tinkerpop.gremlin.structure.Vertex; import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph; import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerStorageGraph; import org.junit.Test; +import java.io.DataInputStream; import java.io.File; import java.io.RandomAccessFile; import java.nio.file.Files; @@ -105,6 +107,55 @@ public class GraphBinaryStorageTest extends AbstractTinkerStorageConformanceTest graph.close(); } + @Test + public void shouldStreamSnapshotAsOneFramePerElement() throws Exception { + final TinkerStorageGraph graph = open(); + final String location = graph.configuration().getString(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION); + final Vertex a = graph.addVertex(T.id, 1, "name", "a"); + final Vertex b = graph.addVertex(T.id, 2, "name", "b"); + final Vertex c = graph.addVertex(T.id, 3, "name", "c"); + a.addEdge("knows", b, T.id, 10); + b.addEdge("knows", c, T.id, 11); + graph.tx().commit(); + graph.compact(); + + // the snapshot must be written as one framed record per element (3 vertices + 2 edges = 5), rather than a + // single whole-graph frame, so compaction never buffers the entire graph in one array + final File snapshotFile = new File(location, GraphBinaryStorage.SNAPSHOT_FILE); + assertEquals(5, countFrames(snapshotFile)); + graph.close(); + + // and the streamed snapshot must reopen to exactly the same graph + final TinkerStorageGraph reopened = open(); + assertEquals(3, countOf(reopened.vertices())); + assertEquals(2, countOf(reopened.edges())); + assertEquals("a", reopened.vertices(1).next().value("name")); + assertEquals("knows", reopened.edges(10).next().label()); + reopened.close(); + } + + /** + * Count the length-prefixed frames in a storage file: each frame is a 4-byte big-endian length followed by that + * many payload bytes. + */ + private static int countFrames(final File file) throws Exception { + int frames = 0; + try (final DataInputStream in = new DataInputStream(new java.io.BufferedInputStream(new java.io.FileInputStream(file)))) { + while (true) { + final int len; + try { + len = in.readInt(); + } catch (java.io.EOFException eof) { + break; + } + final long skipped = in.skip(len); + if (skipped < len) break; + frames++; + } + } + return frames; + } + @Test public void shouldAutoCompactWhenLogExceedsThreshold() { // a small threshold makes automatic compaction fire mid-run, without any explicit compact()/close()
