This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch csr
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/csr by this push:
new aa95963fd0 updates to snapshot
aa95963fd0 is described below
commit aa95963fd05ddc11b13b0eac84fae458cdf4038f
Author: Ken Hu <[email protected]>
AuthorDate: Wed Oct 7 11:12:05 2026 -0700
updates to snapshot
---
.../tinkerpop/gremlin/jsr223/CoreImports.java | 2 +
.../structure/snapshot/build/BuildBudget.java | 132 ++++++++
.../structure/snapshot/build/BuildOptions.java | 3 +-
.../structure/snapshot/build/BuildStats.java | 34 ++
.../structure/snapshot/build/FixedSpool.java | 244 +++++++++++++++
.../snapshot/build/HybridSnapshotBuilder.java | 48 +++
.../gremlin/structure/snapshot/build/IntTable.java | 100 ++++++
.../structure/snapshot/build/PropertySpool.java | 127 +++++++-
.../snapshot/build/PropertySpoolReader.java | 74 ++++-
.../snapshot/build/StreamingSnapshotBuilder.java | 346 ++++++++++++++++-----
spark-gremlin/src/test/csr-memory/csrbench.py | 118 ++++++-
spark-gremlin/src/test/csr-memory/matrix.json | 20 +-
.../tinkerpop/gremlin/spark/csr/CsrMemoryRun.java | 38 ++-
.../structure/CsrSnapshotBuildBenchmark.java | 102 +++---
14 files changed, 1238 insertions(+), 150 deletions(-)
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java
index d398fbf0e1..2f3f0c4d8f 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java
@@ -145,6 +145,7 @@ import
org.apache.tinkerpop.gremlin.structure.io.gryo.GryoWriter;
import org.apache.tinkerpop.gremlin.structure.snapshot.CsrSnapshot;
import org.apache.tinkerpop.gremlin.structure.snapshot.build.BuildOptions;
import
org.apache.tinkerpop.gremlin.structure.snapshot.build.HeapSnapshotBuilder;
+import
org.apache.tinkerpop.gremlin.structure.snapshot.build.HybridSnapshotBuilder;
import
org.apache.tinkerpop.gremlin.structure.snapshot.build.StreamingSnapshotBuilder;
import org.apache.tinkerpop.gremlin.structure.snapshot.format.SnapshotLayout;
import org.apache.tinkerpop.gremlin.structure.snapshot.graph.CsrGraph;
@@ -231,6 +232,7 @@ public final class CoreImports {
CLASS_IMPORTS.add(CsrSnapshot.class);
CLASS_IMPORTS.add(StreamingSnapshotBuilder.class);
CLASS_IMPORTS.add(HeapSnapshotBuilder.class);
+ CLASS_IMPORTS.add(HybridSnapshotBuilder.class);
CLASS_IMPORTS.add(BuildOptions.class);
CLASS_IMPORTS.add(SnapshotLayout.class);
CLASS_IMPORTS.add(CsrNative.class);
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildBudget.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildBudget.java
new file mode 100644
index 0000000000..02ec40e3b4
--- /dev/null
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildBudget.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.structure.snapshot.build;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The byte account of the hybrid builder. Structures that can live in heap
reserve their bytes here before they
+ * allocate them. When a reservation does not fit, the largest {@link
Spillable} structure is moved to scratch files,
+ * and again until the reservation fits or nothing is left to spill. A budget
of zero bytes grants nothing, which makes
+ * every structure behave as in the streaming builder.
+ * <p/>
+ * Not thread-safe; a build is single-threaded.
+ */
+final class BuildBudget {
+
+ /**
+ * A structure that holds budgeted heap and can move its contents to a
scratch file.
+ */
+ interface Spillable {
+ /**
+ * The budgeted bytes currently held in heap.
+ */
+ long heldBytes();
+
+ /**
+ * Moves the contents to scratch and releases the held bytes to the
budget. Does nothing if nothing is held.
+ */
+ void spill();
+ }
+
+ private static final BuildBudget NONE = new BuildBudget(0);
+
+ private final long total;
+ private final List<Spillable> spillables = new ArrayList<>();
+ private long used;
+ private long peak;
+
+ BuildBudget(final long total) {
+ this.total = total;
+ }
+
+ /**
+ * A budget that grants no reservation.
+ */
+ static BuildBudget none() {
+ return NONE;
+ }
+
+ long total() {
+ return total;
+ }
+
+ long used() {
+ return used;
+ }
+
+ long available() {
+ return total - used;
+ }
+
+ /**
+ * The highest number of bytes that were reserved at one time.
+ */
+ long peak() {
+ return peak;
+ }
+
+ void register(final Spillable spillable) {
+ if (total > 0) spillables.add(spillable);
+ }
+
+ void unregister(final Spillable spillable) {
+ spillables.remove(spillable);
+ }
+
+ /**
+ * Reserves bytes, spilling the largest registered structures while the
reservation does not fit.
+ *
+ * @param bytes the number of bytes wanted
+ * @param requester the structure that asks, if it is spillable; when it
is the largest it spills itself and the
+ * reservation is refused
+ * @return whether the bytes are reserved
+ */
+ boolean reserve(final long bytes, final Spillable requester) {
+ if (bytes <= 0) return true;
+ if (bytes > total) return false;
+ while (total - used < bytes) {
+ Spillable victim = null;
+ long largest = 0;
+ for (final Spillable s : spillables) {
+ final long held = s.heldBytes();
+ if (held > largest) {
+ largest = held;
+ victim = s;
+ }
+ }
+ if (victim == null) return false;
+ victim.spill();
+ if (victim == requester) return false;
+ }
+ used += bytes;
+ if (used > peak) peak = used;
+ return true;
+ }
+
+ boolean reserve(final long bytes) {
+ return reserve(bytes, null);
+ }
+
+ void release(final long bytes) {
+ used -= bytes;
+ if (used < 0) throw new IllegalStateException("Released more than
reserved");
+ }
+}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildOptions.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildOptions.java
index e4122d6065..251cb302de 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildOptions.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildOptions.java
@@ -61,7 +61,8 @@ public final class BuildOptions {
}
/**
- * Bytes the streaming builder may use for sort chunks and buffers.
Defaults to
+ * Bytes the streaming builder may use for sort chunks and buffers, and
the heap that the hybrid builder may fill with
+ * spools, lookup maps, sort arrays and adjacency arrays before it spills
to scratch. Defaults to
* {@link #DEFAULT_MEMORY_BUDGET_BYTES}.
*/
public long memoryBudgetBytes() {
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildStats.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildStats.java
index 4b77e03982..0d612adba8 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildStats.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/BuildStats.java
@@ -39,14 +39,31 @@ public final class BuildStats {
private final List<Phase> phases;
private final Map<String, Long> segmentBytes;
+ private final Map<String, Long> timers;
+ private final long peakBudgetBytes;
/**
* @param phases phases in execution order
* @param segmentBytes published size in bytes of each segment, keyed by
relative path, in ascending path order
*/
public BuildStats(final List<Phase> phases, final Map<String, Long>
segmentBytes) {
+ this(phases, segmentBytes, java.util.Map.of(), 0);
+ }
+
+ /**
+ * As above, with sub-timers inside phases and the builder's peak budgeted
heap.
+ *
+ * @param timers nanoseconds spent in parts of a phase, keyed by
a name such as {@code edge-scan.resolve}, in
+ * the order given; they overlap the phases and
each other as their names document
+ * @param peakBudgetBytes the most heap the builder reserved from its
memory budget at one time, 0 when it does not
+ * account for heap
+ */
+ public BuildStats(final List<Phase> phases, final Map<String, Long>
segmentBytes, final Map<String, Long> timers,
+ final long peakBudgetBytes) {
this.phases = List.copyOf(phases);
this.segmentBytes = java.util.Collections.unmodifiableMap(new
java.util.LinkedHashMap<>(segmentBytes));
+ this.timers = java.util.Collections.unmodifiableMap(new
java.util.LinkedHashMap<>(timers));
+ this.peakBudgetBytes = peakBudgetBytes;
}
public List<Phase> phases() {
@@ -57,6 +74,23 @@ public final class BuildStats {
return segmentBytes;
}
+ /**
+ * Nanoseconds spent in parts of a phase, for builders that measure them:
{@code edge-scan.source} is the time spent in
+ * the source between edges, {@code edge-scan.callback} the time in the
builder's per-edge work, and
+ * {@code edge-scan.resolve} the part of that spent resolving endpoints to
ordinals. Empty if none were measured.
+ */
+ public Map<String, Long> timers() {
+ return timers;
+ }
+
+ /**
+ * The most heap the builder reserved from its memory budget at one time,
or 0 when the builder does not account for
+ * heap.
+ */
+ public long peakBudgetBytes() {
+ return peakBudgetBytes;
+ }
+
/**
* The largest {@link Phase#diskBytes()} across all phases, or 0 if there
were none.
*/
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/FixedSpool.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/FixedSpool.java
new file mode 100644
index 0000000000..bb33c832fb
--- /dev/null
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/FixedSpool.java
@@ -0,0 +1,244 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.structure.snapshot.build;
+
+import org.apache.tinkerpop.gremlin.structure.snapshot.format.SegmentReader;
+import org.apache.tinkerpop.gremlin.structure.snapshot.format.SegmentWriter;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A scratch spool of int32 or int64 values, the hybrid builder's counterpart
of a scratch {@link SegmentWriter}. While
+ * the budget grants it chunks the values are held in heap and nothing is
written; once it does not, or when the budget
+ * evicts the spool as its largest holder, the values move to a segment file
for good. Written once and read back
+ * sequentially.
+ */
+final class FixedSpool implements AutoCloseable, BuildBudget.Spillable {
+
+ private static final int CHUNK_BYTES = 64 * 1024;
+
+ private final Path path;
+ private final int width;
+ private final int ioBuffer;
+ private final BuildBudget budget;
+ private final int perChunk;
+
+ private boolean inMemory;
+ private List<int[]> intChunks;
+ private List<long[]> longChunks;
+ private long count;
+ private long heldBytes;
+ private SegmentWriter writer;
+ private boolean finished;
+
+ /**
+ * @param width 4 or 8
+ */
+ FixedSpool(final Path path, final int width, final int ioBuffer, final
BuildBudget budget) {
+ if (width != 4 && width != 8) throw new
IllegalArgumentException("Unsupported width " + width);
+ this.path = path;
+ this.width = width;
+ this.ioBuffer = ioBuffer;
+ this.budget = budget;
+ this.perChunk = CHUNK_BYTES / width;
+ this.inMemory = budget.total() > 0;
+ if (inMemory) {
+ if (width == 4) intChunks = new ArrayList<>();
+ else longChunks = new ArrayList<>();
+ budget.register(this);
+ } else {
+ writer = SegmentWriter.create(path, width, ioBuffer);
+ }
+ }
+
+ Path path() {
+ return path;
+ }
+
+ boolean isInMemory() {
+ return inMemory;
+ }
+
+ long count() {
+ return inMemory ? count : writer.count();
+ }
+
+ @Override
+ public long heldBytes() {
+ return heldBytes;
+ }
+
+ void writeInt(final int value) {
+ if (inMemory) {
+ final int slot = (int) (count % perChunk);
+ if (slot == 0 && !grow()) {
+ writer.writeInt(value);
+ return;
+ }
+ intChunks.get(intChunks.size() - 1)[slot] = value;
+ count++;
+ return;
+ }
+ writer.writeInt(value);
+ }
+
+ void writeLong(final long value) {
+ if (inMemory) {
+ final int slot = (int) (count % perChunk);
+ if (slot == 0 && !grow()) {
+ writer.writeLong(value);
+ return;
+ }
+ longChunks.get(longChunks.size() - 1)[slot] = value;
+ count++;
+ return;
+ }
+ writer.writeLong(value);
+ }
+
+ // adds a chunk; false when the spool had to move to its file instead, in
which case the value goes there
+ private boolean grow() {
+ if (!budget.reserve(CHUNK_BYTES, this)) {
+ spill();
+ return false;
+ }
+ heldBytes += CHUNK_BYTES;
+ if (width == 4) intChunks.add(new int[perChunk]);
+ else longChunks.add(new long[perChunk]);
+ return true;
+ }
+
+ @Override
+ public void spill() {
+ if (!inMemory) return;
+ inMemory = false;
+ budget.unregister(this);
+ writer = SegmentWriter.create(path, width, ioBuffer);
+ if (width == 4) {
+ long remaining = count;
+ for (final int[] chunk : intChunks) {
+ final int n = (int) Math.min(perChunk, remaining);
+ for (int i = 0; i < n; i++) writer.writeInt(chunk[i]);
+ remaining -= n;
+ }
+ } else {
+ long remaining = count;
+ for (final long[] chunk : longChunks) {
+ final int n = (int) Math.min(perChunk, remaining);
+ for (int i = 0; i < n; i++) writer.writeLong(chunk[i]);
+ remaining -= n;
+ }
+ }
+ intChunks = null;
+ longChunks = null;
+ count = 0;
+ budget.release(heldBytes);
+ heldBytes = 0;
+ if (finished) writer.finish();
+ }
+
+ void finish() {
+ if (finished) return;
+ finished = true;
+ if (!inMemory) writer.finish();
+ }
+
+ /**
+ * Opens the finished spool for sequential reading.
+ */
+ Reader reader(final int readBufferBytes) {
+ if (!finished) throw new IllegalStateException("Spool " + path + " is
not finished");
+ return inMemory ? new Reader(this) : new
Reader(SegmentReader.open(path, readBufferBytes));
+ }
+
+ /**
+ * The path of the spool as a finished segment file, writing the values
out first if they are held in heap.
+ */
+ Path materialize() {
+ if (!finished) throw new IllegalStateException("Spool " + path + " is
not finished");
+ spill();
+ return path;
+ }
+
+ /**
+ * Frees the heap the spool holds and returns it to the budget.
+ */
+ void release() {
+ if (!inMemory) return;
+ inMemory = false;
+ budget.unregister(this);
+ intChunks = null;
+ longChunks = null;
+ count = 0;
+ budget.release(heldBytes);
+ heldBytes = 0;
+ }
+
+ @Override
+ public void close() {
+ if (writer != null) writer.close();
+ release();
+ }
+
+ /**
+ * A sequential reader over a finished spool.
+ */
+ static final class Reader implements AutoCloseable {
+ private final SegmentReader file;
+ private final List<int[]> ints;
+ private final List<long[]> longs;
+ private final int perChunk;
+ private long position;
+
+ private Reader(final SegmentReader file) {
+ this.file = file;
+ this.ints = null;
+ this.longs = null;
+ this.perChunk = 0;
+ }
+
+ private Reader(final FixedSpool spool) {
+ this.file = null;
+ this.ints = spool.intChunks;
+ this.longs = spool.longChunks;
+ this.perChunk = spool.perChunk;
+ }
+
+ int readInt() {
+ if (file != null) return file.readInt();
+ final int value = ints.get((int) (position / perChunk))[(int)
(position % perChunk)];
+ position++;
+ return value;
+ }
+
+ long readLong() {
+ if (file != null) return file.readLong();
+ final long value = longs.get((int) (position / perChunk))[(int)
(position % perChunk)];
+ position++;
+ return value;
+ }
+
+ @Override
+ public void close() {
+ if (file != null) file.close();
+ }
+ }
+}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/HybridSnapshotBuilder.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/HybridSnapshotBuilder.java
new file mode 100644
index 0000000000..a10f79813a
--- /dev/null
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/HybridSnapshotBuilder.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.structure.snapshot.build;
+
+import org.apache.tinkerpop.gremlin.structure.snapshot.spi.SnapshotSource;
+
+import java.nio.file.Path;
+import java.util.Objects;
+
+/**
+ * Builds a snapshot bundle with the construction steps of the {@link
StreamingSnapshotBuilder}, but keeps each
+ * structure in heap for as long as {@link BuildOptions#memoryBudgetBytes()}
allows and only spills to scratch files
+ * when the budget runs low. In heap are, in order of preference, the
identifier lookup map used to resolve edge
+ * endpoints, the degree arrays and fill cursors, the spools of identifiers,
labels, index keys and properties, the sort
+ * of the index keys, and the adjacency arrays that are filled before they are
written sequentially. When a reservation
+ * does not fit the largest spool in heap is moved to its file first; a
structure that still does not fit takes the
+ * streaming builder's route (external sort and mapped index, mapped arrays,
file spools). With a budget too small for
+ * anything the build is the streaming build.
+ * <p/>
+ * The budget counts the large structures only; the dictionaries, buffers and
the objects of the source are extra. The
+ * output is byte-for-byte that of the heap and streaming builders for the
same source scan order.
+ */
+public final class HybridSnapshotBuilder implements SnapshotBuilder {
+
+ @Override
+ public BuildStats build(final SnapshotSource source, final Path target,
final BuildOptions options) {
+ Objects.requireNonNull(source);
+ Objects.requireNonNull(target);
+ Objects.requireNonNull(options);
+ return new StreamingSnapshotBuilder(true).build(source, target,
options);
+ }
+}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/IntTable.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/IntTable.java
new file mode 100644
index 0000000000..1df2e67787
--- /dev/null
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/IntTable.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.structure.snapshot.build;
+
+import org.apache.tinkerpop.gremlin.structure.snapshot.format.MappedSegment;
+import org.apache.tinkerpop.gremlin.structure.snapshot.format.Manifest;
+import org.apache.tinkerpop.gremlin.structure.snapshot.format.SegmentWriter;
+
+import java.nio.file.Path;
+
+/**
+ * A zero-initialized int32 array that is either an {@code int[]} in heap or a
memory-mapped scratch or output segment.
+ * The hybrid builder allocates degree arrays, fill cursors and adjacency
arrays in heap while the budget allows it.
+ */
+final class IntTable implements AutoCloseable {
+
+ private final int[] heap;
+ private final MappedSegment mapped;
+ private final long count;
+ private final BuildBudget budget;
+ private final long heldBytes;
+ private boolean closed;
+
+ private IntTable(final int[] heap, final MappedSegment mapped, final long
count, final BuildBudget budget,
+ final long heldBytes) {
+ this.heap = heap;
+ this.mapped = mapped;
+ this.count = count;
+ this.budget = budget;
+ this.heldBytes = heldBytes;
+ }
+
+ /**
+ * A table in heap when the budget grants its bytes, else a mapped file at
the given path.
+ */
+ static IntTable allocate(final BuildBudget budget, final long count, final
Path mappedPath) {
+ final long bytes = count * Integer.BYTES;
+ if (budget.total() > 0 && count <= Integer.MAX_VALUE - 8 &&
budget.reserve(bytes)) {
+ return new IntTable(new int[(int) count], null, count, budget,
bytes);
+ }
+ return new IntTable(null, MappedSegment.create(mappedPath,
Integer.BYTES, count), count, budget, 0);
+ }
+
+ boolean isHeap() {
+ return heap != null;
+ }
+
+ long count() {
+ return count;
+ }
+
+ int get(final long index) {
+ return heap != null ? heap[(int) index] : mapped.getInt(index);
+ }
+
+ void put(final long index, final int value) {
+ if (heap != null) heap[(int) index] = value;
+ else mapped.putInt(index, value);
+ }
+
+ /**
+ * Completes the table as the segment at {@code relativePath}: a mapped
table is finished in place, a table in heap
+ * is written sequentially to the path. Either way the bytes are those of
the heap builder's segment.
+ */
+ Manifest.SegmentInfo publish(final Path path, final String relativePath,
final int ioBuffer) {
+ if (heap == null) {
+ mapped.finish();
+ return mapped.info(relativePath);
+ }
+ try (SegmentWriter out = SegmentWriter.create(path, Integer.BYTES,
ioBuffer)) {
+ for (int i = 0; i < heap.length; i++) out.writeInt(heap[i]);
+ out.finish();
+ return out.info(relativePath);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (closed) return;
+ closed = true;
+ if (mapped != null) mapped.close();
+ if (heldBytes > 0) budget.release(heldBytes);
+ }
+}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpool.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpool.java
index 890d2ae8d4..a8f213c5de 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpool.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpool.java
@@ -26,6 +26,8 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
import java.util.zip.CRC32C;
/**
@@ -41,11 +43,21 @@ import java.util.zip.CRC32C;
* <p/>
* The spool is not thread-safe. I/O failures are reported as {@link
UncheckedIOException}.
*/
-final class PropertySpool implements AutoCloseable {
+final class PropertySpool implements AutoCloseable, BuildBudget.Spillable {
+
+ // the unit in which a spool held in heap reserves budget
+ static final int CHUNK_BYTES = 64 * 1024;
private final Path path;
- private final byte[] buffer;
+ private final int bufferBytes;
+ private byte[] buffer;
private final boolean keepOpen;
+ private final BuildBudget budget;
+ // while in heap, the records are in chunks and nothing is written to the
file
+ private boolean inMemory;
+ private List<byte[]> chunks;
+ private long memoryBytes;
+ private long heldBytes;
private final CRC32C crc = new CRC32C();
private FileChannel channel;
private int position;
@@ -60,9 +72,112 @@ final class PropertySpool implements AutoCloseable {
* @param keepOpen whether to keep the file open until {@link
#finish()} instead of for each flush only
*/
PropertySpool(final Path path, final int bufferBytes, final boolean
keepOpen) {
+ this(path, bufferBytes, keepOpen, BuildBudget.none());
+ }
+
+ /**
+ * A spool that stays in heap while the budget grants it chunks and moves
to the file for good once it does not, or
+ * when the budget evicts it as the largest holder.
+ *
+ * @param budget the budget to reserve chunks from; with a budget that
grants nothing the spool is file-backed from
+ * the start
+ */
+ PropertySpool(final Path path, final int bufferBytes, final boolean
keepOpen, final BuildBudget budget) {
this.path = path;
- this.buffer = new byte[bufferBytes];
+ this.bufferBytes = bufferBytes;
this.keepOpen = keepOpen;
+ this.budget = budget;
+ this.inMemory = budget.total() > 0;
+ if (inMemory) {
+ this.chunks = new ArrayList<>();
+ budget.register(this);
+ } else {
+ this.buffer = new byte[bufferBytes];
+ }
+ }
+
+ /**
+ * Whether the records are held in heap.
+ */
+ boolean isInMemory() {
+ return inMemory;
+ }
+
+ @Override
+ public long heldBytes() {
+ return heldBytes;
+ }
+
+ /**
+ * Writes the records held in heap to the file and continues there.
+ */
+ @Override
+ public void spill() {
+ if (!inMemory) return;
+ inMemory = false;
+ final boolean wasFinished = finished;
+ finished = false;
+ buffer = new byte[bufferBytes];
+ position = 0;
+ final List<byte[]> held = chunks;
+ final long bytes = memoryBytes;
+ chunks = null;
+ memoryBytes = 0;
+ budget.unregister(this);
+ long remaining = bytes;
+ for (final byte[] chunk : held) {
+ final int n = (int) Math.min(CHUNK_BYTES, remaining);
+ if (n > 0) append(chunk, 0, n);
+ remaining -= n;
+ }
+ budget.release(heldBytes);
+ heldBytes = 0;
+ if (wasFinished) finish();
+ }
+
+ /**
+ * Opens the finished spool for reading, from heap when it is there.
+ */
+ PropertySpoolReader reader(final int readBufferBytes) {
+ if (!finished) throw new IllegalStateException("Spool " + path + " is
not finished");
+ return inMemory ? PropertySpoolReader.ofMemory(chunks, CHUNK_BYTES,
memoryBytes)
+ : new PropertySpoolReader(path, readBufferBytes);
+ }
+
+ /**
+ * Frees the heap the spool holds and returns it to the budget. The spool
can no longer be read from heap.
+ */
+ void release() {
+ if (!inMemory) return;
+ inMemory = false;
+ budget.unregister(this);
+ chunks = null;
+ memoryBytes = 0;
+ budget.release(heldBytes);
+ heldBytes = 0;
+ }
+
+ private boolean appendMemory(final byte[] src, final int length) {
+ final long capacity = (long) chunks.size() * CHUNK_BYTES;
+ if (memoryBytes + length > capacity) {
+ final long need = (memoryBytes + length - capacity + CHUNK_BYTES -
1) / CHUNK_BYTES;
+ if (!budget.reserve(need * CHUNK_BYTES, this)) {
+ spill();
+ return false;
+ }
+ heldBytes += need * CHUNK_BYTES;
+ for (long i = 0; i < need; i++) chunks.add(new byte[CHUNK_BYTES]);
+ }
+ int offset = 0;
+ while (offset < length) {
+ final byte[] chunk = chunks.get((int) (memoryBytes / CHUNK_BYTES));
+ final int within = (int) (memoryBytes % CHUNK_BYTES);
+ final int n = Math.min(length - offset, CHUNK_BYTES - within);
+ System.arraycopy(src, offset, chunk, within, n);
+ memoryBytes += n;
+ offset += n;
+ }
+ return true;
}
Path path() {
@@ -81,6 +196,7 @@ final class PropertySpool implements AutoCloseable {
final byte[] src = record.bytes();
final int length = record.length();
records++;
+ if (inMemory && appendMemory(src, length)) return;
if (length > buffer.length - position) {
flush();
if (length >= buffer.length) {
@@ -133,6 +249,10 @@ final class PropertySpool implements AutoCloseable {
*/
void finish() {
if (finished) return;
+ if (inMemory) {
+ finished = true;
+ return;
+ }
flush();
try {
if (!created) {
@@ -159,5 +279,6 @@ final class PropertySpool implements AutoCloseable {
@Override
public void close() {
finish();
+ release();
}
}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpoolReader.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpoolReader.java
index c8f536c914..15360c9c93 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpoolReader.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/PropertySpoolReader.java
@@ -26,6 +26,7 @@ import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.nio.ByteOrder;
import java.nio.file.Path;
+import java.util.List;
/**
* Reads back the records of a finished {@link PropertySpool} sequentially.
The caller knows the record layout, which is
@@ -44,24 +45,77 @@ final class PropertySpoolReader implements AutoCloseable {
int length;
}
+ // a file spool is read through the segment reader; a spool held in heap
is read from its chunks
private final SegmentReader reader;
+ private final List<byte[]> chunks;
+ private final int chunkBytes;
+ private final long length;
+ private long position;
private final byte[] scratch = new byte[8];
PropertySpoolReader(final Path path, final int bufferBytes) {
this.reader = SegmentReader.open(path, bufferBytes);
+ this.chunks = null;
+ this.chunkBytes = 0;
+ this.length = 0;
+ }
+
+ private PropertySpoolReader(final List<byte[]> chunks, final int
chunkBytes, final long length) {
+ this.reader = null;
+ this.chunks = chunks;
+ this.chunkBytes = chunkBytes;
+ this.length = length;
+ }
+
+ /**
+ * A reader over the records of a spool that is held in heap.
+ *
+ * @param chunks the chunks in order, all but the last full
+ * @param chunkBytes the size of a chunk
+ * @param length the number of bytes written
+ */
+ static PropertySpoolReader ofMemory(final List<byte[]> chunks, final int
chunkBytes, final long length) {
+ return new PropertySpoolReader(chunks, chunkBytes, length);
}
boolean hasRemaining() {
- return reader.hasRemaining();
+ return reader != null ? reader.hasRemaining() : position < length;
+ }
+
+ private byte nextByte() {
+ if (reader != null) return reader.readByte();
+ if (position >= length) throw new IllegalStateException("Corrupt
spool: read past the end");
+ final byte b = chunks.get((int) (position / chunkBytes))[(int)
(position % chunkBytes)];
+ position++;
+ return b;
+ }
+
+ private void nextBytes(final byte[] dst, final int offset, final int
count) {
+ if (reader != null) {
+ reader.readBytes(dst, offset, count);
+ return;
+ }
+ if (length - position < count) throw new
IllegalStateException("Corrupt spool: read past the end");
+ int off = offset;
+ int remaining = count;
+ while (remaining > 0) {
+ final byte[] chunk = chunks.get((int) (position / chunkBytes));
+ final int within = (int) (position % chunkBytes);
+ final int n = Math.min(remaining, chunkBytes - within);
+ System.arraycopy(chunk, within, dst, off, n);
+ position += n;
+ off += n;
+ remaining -= n;
+ }
}
int readInt() {
- reader.readBytes(scratch, 0, 4);
+ nextBytes(scratch, 0, 4);
return (int) INT.get(scratch, 0);
}
long readLong() {
- reader.readBytes(scratch, 0, 8);
+ nextBytes(scratch, 0, 8);
return (long) LONG.get(scratch, 0);
}
@@ -70,8 +124,8 @@ final class PropertySpoolReader implements AutoCloseable {
* caller must not call {@link #readPayload} for it.
*/
ValueType readType() {
- final ValueType type = ValueCodec.typeOfCode(reader.readByte());
- if (type == null) throw new IllegalStateException("Corrupt spool " +
reader.path() + ": invalid type code");
+ final ValueType type = ValueCodec.typeOfCode(nextByte());
+ if (type == null) throw new IllegalStateException("Corrupt spool " +
describe() + ": invalid type code");
return type;
}
@@ -80,14 +134,18 @@ final class PropertySpoolReader implements AutoCloseable {
*/
void readPayload(final ValueType type, final Payload into) {
final int length = type.isFixedWidth() ? type.width() : readInt();
- if (length < 0) throw new IllegalStateException("Corrupt spool " +
reader.path() + ": negative length");
+ if (length < 0) throw new IllegalStateException("Corrupt spool " +
describe() + ": negative length");
if (into.bytes.length < length) into.bytes = new byte[Math.max(length,
into.bytes.length * 2)];
- reader.readBytes(into.bytes, 0, length);
+ nextBytes(into.bytes, 0, length);
into.length = length;
}
+ private String describe() {
+ return reader != null ? String.valueOf(reader.path()) : "(in heap)";
+ }
+
@Override
public void close() {
- reader.close();
+ if (reader != null) reader.close();
}
}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/StreamingSnapshotBuilder.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/StreamingSnapshotBuilder.java
index fa3296d6a9..72b165dae5 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/StreamingSnapshotBuilder.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/build/StreamingSnapshotBuilder.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.structure.snapshot.build;
+import com.carrotsearch.hppc.LongIntHashMap;
import org.apache.tinkerpop.gremlin.structure.snapshot.format.BuildDirectory;
import org.apache.tinkerpop.gremlin.structure.snapshot.format.ColumnEncoding;
import org.apache.tinkerpop.gremlin.structure.snapshot.format.ColumnReader;
@@ -108,6 +109,11 @@ import java.util.function.Function;
* </ul>
* Count-hint mismatches, order violations of a grouped scan and edges with
unknown endpoints fail with
* {@link IllegalStateException}. Unsupported data fails with {@link
UnsupportedSnapshotDataException}.
+ * <p/>
+ * A builder created by {@link HybridSnapshotBuilder} runs the same steps with
a storage policy: a byte-accounted
+ * budget of {@link BuildOptions#memoryBudgetBytes()} decides which structures
stay in heap (the spools, the identifier
+ * lookup map, the sort of the index keys, the degree and fill arrays) and
which live in scratch files as described
+ * above. When a reservation does not fit, the largest spool in heap is moved
to its file first. The output is the same.
*/
public final class StreamingSnapshotBuilder implements SnapshotBuilder {
@@ -117,13 +123,23 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private static final String SPOOL_DIR = "spool/";
private static final String DEGREE_DIR = "degree/";
+ private final boolean hybrid;
+
+ public StreamingSnapshotBuilder() {
+ this(false);
+ }
+
+ StreamingSnapshotBuilder(final boolean hybrid) {
+ this.hybrid = hybrid;
+ }
+
@Override
public BuildStats build(final SnapshotSource source, final Path target,
final BuildOptions options) {
Objects.requireNonNull(source);
Objects.requireNonNull(target);
Objects.requireNonNull(options);
try (BuildDirectory dir = BuildDirectory.create(target,
options.scratchDirectory().orElse(null))) {
- return new Build(source, dir, options).run();
+ return new Build(source, dir, options, hybrid).run();
}
}
@@ -167,6 +183,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
* The state of one build.
*/
private static final class Build {
+ private final boolean hybrid;
private final SnapshotSource source;
private final BuildDirectory dir;
private final SnapshotLayout layout;
@@ -181,7 +198,9 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private final long budget;
private final int ioBuffer;
private final int spoolBuffer;
- private final ExternalIndexSorter sorter;
+ private final BuildBudget account;
+ private final ExternalIndexSorter streamingSorter;
+ private final Map<String, Long> timers = new LinkedHashMap<>();
private final List<BuildStats.Phase> phases = new ArrayList<>();
private final List<Manifest.SegmentInfo> segments = new ArrayList<>();
@@ -195,36 +214,47 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private final LabelCounts edgeLabelHistogram = new LabelCounts();
private final LabelDictionary vertexKeys = new LabelDictionary();
private final List<KeyColumn> vertexColumns = new ArrayList<>();
- private SegmentWriter vertexLabelSpool;
- private SegmentWriter vertexLabelCountSpool;
+ private FixedSpool vertexLabelSpool;
+ private FixedSpool vertexLabelCountSpool;
private boolean multiLabels;
private final LabelDictionary metaKeys = new LabelDictionary();
private PropertySpool vertexIdSpool;
private final ColumnStats vertexIdStats = new ColumnStats();
- private SegmentWriter vertexKeySpool;
+ private FixedSpool vertexKeySpool;
private boolean vertexKeysSorted = true;
private long lastVertexKey;
// vertex index
private Manifest.ColumnInfo vertexIdInfo;
private IdentifierIndex vertexIndex;
+ private long mapBytes;
+ // hybrid: the lookup in heap, from index key to the first ordinal
with it, chained through vertexNext for keys
+ // that more than one identifier shares (non-integral identifiers and
mixed integral types)
+ private LongIntHashMap vertexMap;
+ private int[] vertexNext;
+ private ValueType vertexExact;
+ private ColumnReader vertexIds;
// edge scan
private int edgeCount;
private final LabelDictionary edgeLabels = new LabelDictionary();
private final LabelDictionary edgeKeys = new LabelDictionary();
private final List<KeyColumn> edgeColumns = new ArrayList<>();
- private SegmentWriter edgeLabelSpool;
+ private FixedSpool edgeLabelSpool;
private PropertySpool edgeIdSpool;
private final ColumnStats edgeIdStats = new ColumnStats();
- private SegmentWriter edgeKeySpool;
+ private FixedSpool edgeKeySpool;
private boolean edgeKeysSorted = true;
private long lastEdgeKey;
private SegmentWriter outVertices;
private SegmentWriter inVertices;
private SegmentWriter fastOutNeighbors;
- private MappedSegment outCursors;
- private MappedSegment inCursors;
+ private IntTable outCursors;
+ private IntTable inCursors;
+ private long sourceNanos;
+ private long callbackNanos;
+ private long resolveNanos;
+ private long lastExitNanos;
private boolean haveLastOut;
private int lastOutOrdinal;
private ValueType lastOutType;
@@ -242,7 +272,8 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private final VertexProperties vertexProperties = new
VertexProperties();
private final EdgeProperties edgeProperties = new EdgeProperties();
- Build(final SnapshotSource source, final BuildDirectory dir, final
BuildOptions options) {
+ Build(final SnapshotSource source, final BuildDirectory dir, final
BuildOptions options, final boolean hybrid) {
+ this.hybrid = hybrid;
this.source = source;
this.dir = dir;
this.layout = options.layout();
@@ -254,7 +285,14 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
this.budget = options.memoryBudgetBytes();
this.ioBuffer = (int) Math.max(64,
Math.min(SegmentWriter.DEFAULT_BUFFER_BYTES, budget / 64));
this.spoolBuffer = (int) Math.max(256, Math.min(32 * 1024, budget
/ 256));
- this.sorter = new ExternalIndexSorter(dir::scratchPath, budget);
+ this.account = hybrid ? new BuildBudget(budget) :
BuildBudget.none();
+ this.streamingSorter = hybrid ? null : new
ExternalIndexSorter(dir::scratchPath, budget);
+ }
+
+ // the streaming sorter is sized from the whole budget; in hybrid mode
it gets what is left when it is needed
+ private ExternalIndexSorter sorter() {
+ if (!hybrid) return streamingSorter;
+ return new ExternalIndexSorter(dir::scratchPath,
Math.max(account.available(), budget / 4));
}
BuildStats run() {
@@ -294,7 +332,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
dir.publish(manifest);
phases.add(new BuildStats.Phase("publish", System.nanoTime() -
start, diskBytes));
- return new BuildStats(phases, segmentBytes(manifest));
+ return new BuildStats(phases, segmentBytes(manifest), timers,
account.peak());
} finally {
closeAll();
}
@@ -341,11 +379,6 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
if (published) segments.add(writer.info(relativePath));
}
- private void publish(final MappedSegment segment, final String
relativePath) {
- segment.finish();
- segments.add(segment.info(relativePath));
- }
-
private static void checkCount(final String kind, final OptionalLong
hint, final long actual) {
if (hint.isPresent() && hint.getAsLong() != actual) {
throw new IllegalStateException("The source announced " +
hint.getAsLong() + " " + kind
@@ -353,11 +386,19 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
}
}
- private static KeyColumn column(final List<KeyColumn> columns, final
int code, final Path path,
- final int bufferBytes) {
+ private FixedSpool fixed(final String relativePath, final int width) {
+ return track(new FixedSpool(dir.scratchPath(relativePath), width,
ioBuffer, account));
+ }
+
+ private PropertySpool propertySpool(final String relativePath, final
int bufferBytes, final boolean keepOpen) {
+ return new PropertySpool(dir.scratchPath(relativePath),
bufferBytes, keepOpen, account);
+ }
+
+ private KeyColumn column(final List<KeyColumn> columns, final int
code, final String relativePath,
+ final int bufferBytes) {
if (code < columns.size()) return columns.get(code);
if (code != columns.size()) throw new
IllegalStateException("Property-key code " + code + " skipped");
- final KeyColumn column = new KeyColumn(new PropertySpool(path,
bufferBytes, false));
+ final KeyColumn column = new KeyColumn(propertySpool(relativePath,
bufferBytes, false));
columns.add(column);
return column;
}
@@ -366,12 +407,18 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private void scanVertices() {
if (withIdentity) {
- vertexLabelSpool = writer(false, SPOOL_DIR +
"vertex-labels.bin", 4);
- vertexLabelCountSpool = writer(false, SPOOL_DIR +
"vertex-label-counts.bin", 4);
+ vertexLabelSpool = fixed(SPOOL_DIR + "vertex-labels.bin", 4);
+ vertexLabelCountSpool = fixed(SPOOL_DIR +
"vertex-label-counts.bin", 4);
}
- vertexIdSpool = track(new PropertySpool(dir.scratchPath(SPOOL_DIR
+ "vertex-ids.bin"), ioBuffer, true));
- vertexKeySpool = writer(false, SPOOL_DIR +
"vertex-index-keys.bin", 8);
+ vertexIdSpool = track(propertySpool(SPOOL_DIR + "vertex-ids.bin",
ioBuffer, true));
+ vertexKeySpool = fixed(SPOOL_DIR + "vertex-index-keys.bin", 8);
+ lastExitNanos = System.nanoTime();
source.scanVertices(this::onVertex);
+ sourceNanos += System.nanoTime() - lastExitNanos;
+ timers.put("vertex-scan.source", sourceNanos);
+ timers.put("vertex-scan.callback", callbackNanos);
+ sourceNanos = 0;
+ callbackNanos = 0;
checkCount("vertices", source.vertexCount(), vertexCount);
if (withIdentity) {
vertexLabelSpool.finish();
@@ -386,6 +433,14 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
}
private void onVertex(final Object id, final List<String> labels,
final VertexPropertySource properties) {
+ final long enter = System.nanoTime();
+ sourceNanos += enter - lastExitNanos;
+ addVertex(id, labels, properties);
+ lastExitNanos = System.nanoTime();
+ callbackNanos += lastExitNanos - enter;
+ }
+
+ private void addVertex(final Object id, final List<String> labels,
final VertexPropertySource properties) {
if (vertexCount >= MAX_ELEMENTS) {
throw new UnsupportedSnapshotDataException("More than " +
MAX_ELEMENTS + " vertices are not supported");
}
@@ -427,8 +482,8 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
final int code = vertexKeys.codeOf(key);
final ValueType idType = ValueCodec.requireIdentifierType(id,
"vertex property identifier", key);
final ValueType type = value == null ? null :
ValueCodec.requireType(value, "vertex property", key);
- final KeyColumn column = column(vertexColumns, code,
- dir.scratchPath(SPOOL_DIR + "vertex-property-" + code
+ ".bin"), spoolBuffer);
+ final KeyColumn column = column(vertexColumns, code, SPOOL_DIR
+ "vertex-property-" + code + ".bin",
+ spoolBuffer);
if (column.lastOrdinal == ordinal) {
column.multi = true;
} else {
@@ -475,8 +530,8 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
: ValueCodec.requireType(value, "meta-property of '" +
key + "'", metaKey);
MetaColumn meta = column.metas.get(metaCode);
if (meta == null) {
- meta = new MetaColumn(new
PropertySpool(dir.scratchPath(SPOOL_DIR + "vertex-meta-" + code + "-"
- + metaCode + ".bin"), spoolBuffer, false));
+ meta = new MetaColumn(propertySpool(SPOOL_DIR +
"vertex-meta-" + code + "-" + metaCode + ".bin",
+ spoolBuffer, false));
column.metas.put(metaCode, meta);
}
if (meta.lastIndex == index) {
@@ -504,11 +559,55 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
vertexIdInfo = compactIds(SegmentPaths.VERTEX_IDS_DIR,
vertexIdSpool, vertexCount, vertexIdStats);
final ValueType exact = IdentifierIndex.exactType(vertexIdInfo);
final ColumnReader ids = exact == null ?
openIds(SegmentPaths.VERTEX_IDS_DIR, vertexCount, vertexIdInfo) : null;
- writeIndex("vertex", vertexKeySpool.path(), vertexKeysSorted,
vertexCount, vertexIdInfo, ids, withIdentity,
- SegmentPaths.VERTEX_ID_INDEX_KEYS,
SegmentPaths.VERTEX_ID_INDEX_ORDINALS, "vertex");
- final Function<String, MappedSegment> opener =
opener(withIdentity);
- vertexIndex =
IdentifierIndex.of(opener.apply(SegmentPaths.VERTEX_ID_INDEX_KEYS),
- opener.apply(SegmentPaths.VERTEX_ID_INDEX_ORDINALS),
exact, ids);
+ final ExternalIndexSorter.EntrySink tap = hybrid ?
reserveLookupMap(exact, ids) : null;
+ writeIndex("vertex", vertexKeySpool, vertexKeysSorted,
vertexCount, vertexIdInfo, ids, withIdentity,
+ SegmentPaths.VERTEX_ID_INDEX_KEYS,
SegmentPaths.VERTEX_ID_INDEX_ORDINALS, "vertex", tap);
+ vertexKeySpool.release();
+ if (vertexMap == null) {
+ final Function<String, MappedSegment> opener =
opener(withIdentity);
+ vertexIndex =
IdentifierIndex.of(opener.apply(SegmentPaths.VERTEX_ID_INDEX_KEYS),
+ opener.apply(SegmentPaths.VERTEX_ID_INDEX_ORDINALS),
exact, ids);
+ }
+ }
+
+ /**
+ * Hybrid mode: reserves the heap for the lookup map and returns the
sink that fills it from the sorted index
+ * entries, or null when the map does not fit and the mapped index
serves the lookups. Integral identifiers are
+ * keyed by value. Other identifiers are keyed by the 64-bit key hash
and chained, and a candidate is checked
+ * against the encoded identifier in the identifier column.
+ */
+ private ExternalIndexSorter.EntrySink reserveLookupMap(final ValueType
exact, final ColumnReader ids) {
+ final long n = vertexCount;
+ // hppc sizes its arrays to the next power of two of n /
loadFactor, plus a sentinel slot
+ final long capacity = Long.highestOneBit(Math.max(4, (long)
Math.ceil(n / 0.75)) - 1) << 1;
+ if (capacity > (1L << 30)) return null;
+ final long bytes = (capacity + 1) * (Long.BYTES + Integer.BYTES) +
(exact == null ? n * Integer.BYTES : 0);
+ if (!account.reserve(bytes)) return null;
+ mapBytes = bytes;
+ vertexExact = exact;
+ vertexIds = ids;
+ vertexMap = new LongIntHashMap((int) n);
+ if (exact != null) return (key, ordinal) -> vertexMap.put(key,
ordinal);
+ vertexNext = new int[(int) n];
+ return (key, ordinal) -> {
+ vertexNext[ordinal] = vertexMap.getOrDefault(key, -1);
+ vertexMap.put(key, ordinal);
+ };
+ }
+
+ private int lookupVertex(final Object id) {
+ if (vertexMap == null) return vertexIndex.lookup(id);
+ final ValueType type = ValueCodec.typeOf(id);
+ if (type == null) return -1;
+ if (vertexExact != null) {
+ return type != vertexExact ? -1 :
vertexMap.getOrDefault(((Number) id).longValue(), -1);
+ }
+ final byte[] encoded = ValueCodec.encode(type, id);
+ final long key = IdentifierIndex.keyOf(type, encoded, 0,
encoded.length);
+ for (int ordinal = vertexMap.getOrDefault(key, -1); ordinal >= 0;
ordinal = vertexNext[ordinal]) {
+ if (vertexIds.matches(ordinal, type, encoded)) return ordinal;
+ }
+ return -1;
}
/**
@@ -518,7 +617,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
final ColumnStats stats) {
final Manifest.ColumnInfo info = stats.toInfo(count);
try (ColumnWriter writer =
ColumnWriter.create(resolver(withIdentity), columnDir, count, info);
- PropertySpoolReader in = new
PropertySpoolReader(spool.path(), ioBuffer)) {
+ PropertySpoolReader in = spool.reader(ioBuffer)) {
final PropertySpoolReader.Payload payload = new
PropertySpoolReader.Payload();
for (long ordinal = 0; ordinal < count; ordinal++) {
final ValueType type = in.readType();
@@ -528,6 +627,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
final List<Manifest.SegmentInfo> written = writer.finish();
if (withIdentity) segments.addAll(written);
}
+ spool.release();
return info;
}
@@ -539,18 +639,43 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
* Writes an identifier index from the key spool, whose entry
positions are the ordinals, either by copying the
* spool when it is already in index order or by an external sort. The
index is published only when asked.
*/
- private void writeIndex(final String kind, final Path keySpool, final
boolean sorted, final long count,
+ private void writeIndex(final String kind, final FixedSpool keySpool,
final boolean sorted, final long count,
final Manifest.ColumnInfo idInfo, final
ColumnReader ids, final boolean published,
- final String keysPath, final String
ordinalsPath, final String sortName) {
+ final String keysPath, final String
ordinalsPath, final String sortName,
+ final ExternalIndexSorter.EntrySink tap) {
final Function<String, Path> resolver = resolver(published);
try (IdentifierIndexWriter writer =
IdentifierIndexWriter.create(resolver.apply(keysPath),
resolver.apply(ordinalsPath), kind, count,
IdentifierIndex.exactType(idInfo), ids)) {
+ // the entries reach the sink in index order, the writer first
so that a duplicate fails before the tap
+ final ExternalIndexSorter.EntrySink sink = tap == null ?
writer::add : (key, ordinal) -> {
+ writer.add(key, ordinal);
+ tap.accept(key, ordinal);
+ };
+ // the key, the ordinal and the two copies that the stable
in-heap sort needs
+ final long sortBytes = 24L * count;
if (sorted) {
- try (SegmentReader in = SegmentReader.open(keySpool,
ioBuffer)) {
- for (long ordinal = 0; ordinal < count; ordinal++)
writer.add(in.readLong(), (int) ordinal);
+ try (FixedSpool.Reader in = keySpool.reader(ioBuffer)) {
+ for (long ordinal = 0; ordinal < count; ordinal++)
sink.accept(in.readLong(), (int) ordinal);
+ }
+ } else if (hybrid && count <= MAX_ELEMENTS &&
account.reserve(sortBytes)) {
+ try {
+ final int n = (int) count;
+ final long[] keys = new long[n];
+ final int[] ordinals = new int[n];
+ try (FixedSpool.Reader in = keySpool.reader(ioBuffer))
{
+ for (int i = 0; i < n; i++) {
+ keys[i] = in.readLong();
+ ordinals[i] = i;
+ }
+ }
+ keySpool.release();
+ IdentifierIndex.sortEntries(keys, ordinals, n);
+ for (int i = 0; i < n; i++) sink.accept(keys[i],
ordinals[i]);
+ } finally {
+ account.release(sortBytes);
}
} else {
- sorter.sort(keySpool, sortName, writer::add);
+ sorter().sort(keySpool.materialize(), sortName, sink);
}
final List<Manifest.SegmentInfo> written =
writer.finish(keysPath, ordinalsPath);
if (published) segments.addAll(written);
@@ -561,17 +686,28 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private void scanEdges() {
if (withIdentity) {
- edgeLabelSpool = writer(false, SPOOL_DIR + "edge-labels.bin",
4);
- edgeIdSpool = track(new
PropertySpool(dir.scratchPath(SPOOL_DIR + "edge-ids.bin"), ioBuffer, true));
+ edgeLabelSpool = fixed(SPOOL_DIR + "edge-labels.bin", 4);
+ edgeIdSpool = track(propertySpool(SPOOL_DIR + "edge-ids.bin",
ioBuffer, true));
}
- if (edgeIndex) edgeKeySpool = writer(false, SPOOL_DIR +
"edge-index-keys.bin", 8);
+ if (edgeIndex) edgeKeySpool = fixed(SPOOL_DIR +
"edge-index-keys.bin", 8);
outVertices = writer(withIdentity, SegmentPaths.EDGE_OUT_VERTICES,
4);
inVertices = writer(withIdentity, SegmentPaths.EDGE_IN_VERTICES,
4);
if (fastPath) fastOutNeighbors = writer(true,
SegmentPaths.OUT_NEIGHBORS, 4);
- outCursors = track(MappedSegment.create(dir.scratchPath(DEGREE_DIR
+ "out.bin"), 4, vertexCount));
- inCursors = track(MappedSegment.create(dir.scratchPath(DEGREE_DIR
+ "in.bin"), 4, vertexCount));
+ outCursors = track(IntTable.allocate(account, vertexCount,
dir.scratchPath(DEGREE_DIR + "out.bin")));
+ inCursors = track(IntTable.allocate(account, vertexCount,
dir.scratchPath(DEGREE_DIR + "in.bin")));
+ lastExitNanos = System.nanoTime();
source.scanEdges(this::onEdge);
+ sourceNanos += System.nanoTime() - lastExitNanos;
+ timers.put("edge-scan.source", sourceNanos);
+ timers.put("edge-scan.callback", callbackNanos);
+ timers.put("edge-scan.resolve", resolveNanos);
+ if (vertexMap != null) {
+ // the endpoints are resolved; the map is not needed again
+ vertexMap = null;
+ vertexNext = null;
+ account.release(mapBytes);
+ }
checkCount("edges", source.edgeCount(), edgeCount);
if (withProperties) source.scanVariables(this::onVariable);
@@ -588,6 +724,15 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private void onEdge(final Object id, final String label, final Object
outId, final Object inId,
final PropertySource properties) {
+ final long enter = System.nanoTime();
+ sourceNanos += enter - lastExitNanos;
+ addEdge(id, label, outId, inId, properties);
+ lastExitNanos = System.nanoTime();
+ callbackNanos += lastExitNanos - enter;
+ }
+
+ private void addEdge(final Object id, final String label, final Object
outId, final Object inId,
+ final PropertySource properties) {
if (edgeCount >= MAX_ELEMENTS) {
throw new UnsupportedSnapshotDataException("More than " +
MAX_ELEMENTS + " edges are not supported");
}
@@ -609,6 +754,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
}
}
+ final long resolveStart = System.nanoTime();
final int out = resolveOut(outId, ordinal);
if (grouped) {
if (out < lastGroupOrdinal) {
@@ -617,15 +763,16 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
}
lastGroupOrdinal = out;
}
- final int in = vertexIndex.lookup(inId);
+ final int in = lookupVertex(inId);
+ resolveNanos += System.nanoTime() - resolveStart;
if (in < 0) {
throw new IllegalStateException("Edge " + ordinal + " refers
to unknown in-vertex " + inId);
}
outVertices.writeInt(out);
inVertices.writeInt(in);
- outCursors.putInt(out, outCursors.getInt(out) + 1);
- inCursors.putInt(in, inCursors.getInt(in) + 1);
+ outCursors.put(out, outCursors.get(out) + 1);
+ inCursors.put(in, inCursors.get(in) + 1);
if (fastPath) fastOutNeighbors.writeInt(in);
if (withProperties) {
@@ -662,7 +809,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
: id.equals(lastOutId);
if (same) return lastOutOrdinal;
}
- final int ordinal = vertexIndex.lookup(id);
+ final int ordinal = lookupVertex(id);
if (ordinal < 0) {
throw new IllegalStateException("Edge " + edgeOrdinal + "
refers to unknown out-vertex " + id);
}
@@ -685,8 +832,8 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
public void property(final String key, final Object value) {
final int code = edgeKeys.codeOf(key);
final ValueType type = value == null ? null :
ValueCodec.requireType(value, "edge property", key);
- final KeyColumn column = column(edgeColumns, code,
- dir.scratchPath(SPOOL_DIR + "edge-property-" + code +
".bin"), spoolBuffer);
+ final KeyColumn column = column(edgeColumns, code, SPOOL_DIR +
"edge-property-" + code + ".bin",
+ spoolBuffer);
if (column.lastOrdinal == ordinal) {
throw new UnsupportedSnapshotDataException("Edge " +
ordinal + " has more than one value for key '"
+ key + "'");
@@ -719,10 +866,10 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
for (int v = 0; v < vertexCount; v++) {
outOffsets.writeLong(outRunning);
inOffsets.writeLong(inRunning);
- final int outDegree = outCursors.getInt(v);
- final int inDegree = inCursors.getInt(v);
- outCursors.putInt(v, (int) outRunning);
- inCursors.putInt(v, (int) inRunning);
+ final int outDegree = outCursors.get(v);
+ final int inDegree = inCursors.get(v);
+ outCursors.put(v, (int) outRunning);
+ inCursors.put(v, (int) inRunning);
outRunning += outDegree;
inRunning += inDegree;
}
@@ -740,14 +887,10 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
private void replay() {
final boolean outEdges = withIdentity && !grouped;
- final MappedSegment inNeighbors =
track(MappedSegment.create(dir.segmentPath(SegmentPaths.IN_NEIGHBORS), 4,
- edgeCount));
- final MappedSegment inEdges = withIdentity
- ?
track(MappedSegment.create(dir.segmentPath(SegmentPaths.IN_EDGES), 4,
edgeCount)) : null;
- final MappedSegment outNeighbors = fastPath ? null
- :
track(MappedSegment.create(dir.segmentPath(SegmentPaths.OUT_NEIGHBORS), 4,
edgeCount));
- final MappedSegment outEdgeSegment = !fastPath && outEdges
- ?
track(MappedSegment.create(dir.segmentPath(SegmentPaths.OUT_EDGES), 4,
edgeCount)) : null;
+ final IntTable inNeighbors = fill(SegmentPaths.IN_NEIGHBORS);
+ final IntTable inEdges = withIdentity ?
fill(SegmentPaths.IN_EDGES) : null;
+ final IntTable outNeighbors = fastPath ? null :
fill(SegmentPaths.OUT_NEIGHBORS);
+ final IntTable outEdgeSegment = !fastPath && outEdges ?
fill(SegmentPaths.OUT_EDGES) : null;
try (SegmentReader outReader =
SegmentReader.open(resolver(withIdentity).apply(SegmentPaths.EDGE_OUT_VERTICES),
ioBuffer);
@@ -756,25 +899,38 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
for (int edge = 0; edge < edgeCount; edge++) {
final int out = outReader.readInt();
final int in = inReader.readInt();
- final int inPosition = inCursors.getInt(in);
- inCursors.putInt(in, inPosition + 1);
- inNeighbors.putInt(inPosition, out);
- if (inEdges != null) inEdges.putInt(inPosition, edge);
+ final int inPosition = inCursors.get(in);
+ inCursors.put(in, inPosition + 1);
+ inNeighbors.put(inPosition, out);
+ if (inEdges != null) inEdges.put(inPosition, edge);
if (!fastPath) {
- final int outPosition = outCursors.getInt(out);
- outCursors.putInt(out, outPosition + 1);
- outNeighbors.putInt(outPosition, in);
- if (outEdgeSegment != null)
outEdgeSegment.putInt(outPosition, edge);
+ final int outPosition = outCursors.get(out);
+ outCursors.put(out, outPosition + 1);
+ outNeighbors.put(outPosition, in);
+ if (outEdgeSegment != null)
outEdgeSegment.put(outPosition, edge);
}
}
}
+ // the cursors are spent
+ outCursors.close();
+ inCursors.close();
publish(inNeighbors, SegmentPaths.IN_NEIGHBORS);
if (inEdges != null) publish(inEdges, SegmentPaths.IN_EDGES);
if (outNeighbors != null) publish(outNeighbors,
SegmentPaths.OUT_NEIGHBORS);
if (outEdgeSegment != null) publish(outEdgeSegment,
SegmentPaths.OUT_EDGES);
}
+ // an adjacency array of the edge count: in heap while the budget
allows, else the mapped output segment
+ private IntTable fill(final String relativePath) {
+ return track(IntTable.allocate(account, edgeCount,
dir.segmentPath(relativePath)));
+ }
+
+ private void publish(final IntTable table, final String relativePath) {
+ segments.add(table.publish(dir.segmentPath(relativePath),
relativePath, ioBuffer));
+ table.close();
+ }
+
// ------------------------------------------------------------ step
6: compaction
private final List<Manifest.ColumnInfo> vertexValueInfos = new
ArrayList<>();
@@ -790,11 +946,12 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
final ValueType exact =
IdentifierIndex.exactType(edgeIdInfo);
final ColumnReader ids = exact == null
? openIds(SegmentPaths.EDGE_IDS_DIR, edgeCount,
edgeIdInfo) : null;
- writeIndex("edge", edgeKeySpool.path(), edgeKeysSorted,
edgeCount, edgeIdInfo, ids, true,
- SegmentPaths.EDGE_ID_INDEX_KEYS,
SegmentPaths.EDGE_ID_INDEX_ORDINALS, "edge");
+ writeIndex("edge", edgeKeySpool, edgeKeysSorted,
edgeCount, edgeIdInfo, ids, true,
+ SegmentPaths.EDGE_ID_INDEX_KEYS,
SegmentPaths.EDGE_ID_INDEX_ORDINALS, "edge", null);
+ edgeKeySpool.release();
}
- segments.add(LabelDictionary.narrow(edgeLabelSpool.path(),
dir.segmentPath(SegmentPaths.EDGE_LABELS),
- SegmentPaths.EDGE_LABELS, edgeLabels.size()));
+ segments.add(narrow(edgeLabelSpool, SegmentPaths.EDGE_LABELS,
edgeLabels.size()));
+ edgeLabelSpool.release();
}
if (withProperties) {
compactVertexProperties();
@@ -809,13 +966,14 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
*/
private void compactVertexLabels() {
if (!multiLabels) {
- segments.add(LabelDictionary.narrow(vertexLabelSpool.path(),
- dir.segmentPath(SegmentPaths.VERTEX_LABELS),
SegmentPaths.VERTEX_LABELS, vertexLabels.size()));
+ segments.add(narrow(vertexLabelSpool,
SegmentPaths.VERTEX_LABELS, vertexLabels.size()));
+ vertexLabelSpool.release();
+ vertexLabelCountSpool.release();
return;
}
try (LabelListWriter writer =
LabelListWriter.create(dir::segmentPath, vertexLabels.size());
- SegmentReader codes =
SegmentReader.open(vertexLabelSpool.path(), ioBuffer);
- SegmentReader counts =
SegmentReader.open(vertexLabelCountSpool.path(), ioBuffer)) {
+ FixedSpool.Reader codes = vertexLabelSpool.reader(ioBuffer);
+ FixedSpool.Reader counts =
vertexLabelCountSpool.reader(ioBuffer)) {
for (int v = 0; v < vertexCount; v++) {
final int count = counts.readInt();
for (int i = 0; i < count; i++)
writer.add(codes.readInt());
@@ -823,6 +981,29 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
}
segments.addAll(writer.finish(vertexCount));
}
+ vertexLabelSpool.release();
+ vertexLabelCountSpool.release();
+ }
+
+ /**
+ * Converts a spool of int32 label codes into a {@code labels.bin} of
the narrowest width for the final
+ * dictionary, as {@link LabelDictionary#narrow}, reading from heap
when the spool is there.
+ */
+ private Manifest.SegmentInfo narrow(final FixedSpool spool, final
String relativePath, final int labelCount) {
+ try (FixedSpool.Reader in = spool.reader(ioBuffer);
+ SegmentWriter out =
LabelDictionary.createCodeWriter(dir.segmentPath(relativePath), labelCount)) {
+ final long n = spool.count();
+ for (long i = 0; i < n; i++) {
+ final int code = in.readInt();
+ if (code < 0 || code >= labelCount) {
+ throw new IllegalStateException("Label code " + code +
" at index " + i + " of " + relativePath
+ + " is outside the dictionary of " +
labelCount + " labels");
+ }
+ LabelDictionary.writeCode(out, code);
+ }
+ out.finish();
+ return out.info(relativePath);
+ }
}
private void compactVertexProperties() {
@@ -839,7 +1020,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
elementCount, valueInfo, idInfo);
OwnerWriter owners = multi ?
OwnerWriter.create(dir::segmentPath, code, vertexCount,
column.ownerCount) : null;
- PropertySpoolReader in = new
PropertySpoolReader(column.spool.path(), ioBuffer)) {
+ PropertySpoolReader in = column.spool.reader(ioBuffer)) {
long index = 0;
int owner = -1;
long ownerProperties = 0;
@@ -870,6 +1051,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
segments.addAll(owners.finish(column.propertyCount));
}
}
+ column.spool.release();
vertexValueInfos.add(valueInfo);
vertexIdColumnInfos.add(idInfo);
@@ -896,7 +1078,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
final Manifest.ColumnInfo info = meta.values.toInfo(elementCount);
try (ColumnWriter writer = ColumnWriter.create(dir::segmentPath,
SegmentPaths.vertexPropertyMetaDir(keyCode, metaCode),
elementCount, info);
- PropertySpoolReader in = new
PropertySpoolReader(meta.spool.path(), ioBuffer)) {
+ PropertySpoolReader in = meta.spool.reader(ioBuffer)) {
while (in.hasRemaining()) {
final long index = in.readLong();
final int ordinal = in.readInt();
@@ -911,6 +1093,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
}
segments.addAll(writer.finish());
}
+ meta.spool.release();
return info;
}
@@ -921,7 +1104,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
final Manifest.ColumnInfo info =
column.values.toInfo(edgeCount);
try (ColumnWriter writer =
ColumnWriter.create(dir::segmentPath, SegmentPaths.edgePropertyDir(code),
edgeCount, info);
- PropertySpoolReader in = new
PropertySpoolReader(column.spool.path(), ioBuffer)) {
+ PropertySpoolReader in = column.spool.reader(ioBuffer)) {
while (in.hasRemaining()) {
final int ordinal = in.readInt();
final ValueType type = in.readType();
@@ -934,6 +1117,7 @@ public final class StreamingSnapshotBuilder implements
SnapshotBuilder {
}
segments.addAll(writer.finish());
}
+ column.spool.release();
edgeValueInfos.add(info);
}
}
diff --git a/spark-gremlin/src/test/csr-memory/csrbench.py
b/spark-gremlin/src/test/csr-memory/csrbench.py
index df529f95c6..357d5a195e 100755
--- a/spark-gremlin/src/test/csr-memory/csrbench.py
+++ b/spark-gremlin/src/test/csr-memory/csrbench.py
@@ -26,6 +26,7 @@ library only. See README.md in this directory.
import argparse
import csv
+import filecmp
import datetime
import json
import os
@@ -161,9 +162,15 @@ def expand_matrix(matrix, only=None):
if mode == "build":
builder = e.get("builder", "streaming")
source = e.get("source", "gryo")
- budget = int(e.get("builderBudget", 256 * MB))
- r = dict(base, system=builder, query="b%s-%s" %
(human_bytes(budget), source),
- params={"builder": builder, "source": source,
"builderBudget": budget})
+ # budgetOfPeak: a fraction of the peak budget use of the latest
unconstrained hybrid build of the same
+ # dataset, resolved when the run starts (see
resolve_relative_budget)
+ of_peak = e.get("budgetOfPeak")
+ budget = None if of_peak is not None else
int(e.get("builderBudget", 256 * MB))
+ label = "%dpct" % round(float(of_peak) * 100) if of_peak is not
None else human_bytes(budget)
+ query = "b%s-%s" % (label, source) + ("-" + e["tag"] if
e.get("tag") else "")
+ r = dict(base, system=builder, query=query,
+ params={"builder": builder, "source": source,
"builderBudget": budget,
+ "budgetOfPeak": of_peak, "tag": e.get("tag")})
runs.append(r)
elif mode == "open":
systems = e.get("systems") or [e["system"]]
@@ -225,7 +232,8 @@ def build_command(run, matrix, run_dir, work_dir,
classpath, cgroup_dir):
if mode == "build":
a += ["--dataset", run["dataset"], "--snapshot", run["snapshot"],
"--builder", run["params"]["builder"], "--source",
run["params"]["source"],
- "--builder-budget", str(run["params"]["builderBudget"])]
+ "--builder-budget", str(run["params"]["builderBudget"])
+ if run["params"]["builderBudget"] is not None else
"<budgetOfPeak>"]
else:
a += ["--system", system]
if system in CSR_SYSTEMS:
@@ -638,8 +646,54 @@ def unsafe_to_delete(snapshot, run, matrix):
return False
+def compare_trees(reference, other):
+ """Byte-for-byte comparison of two snapshot directories: identical when
both hold the same relative file paths
+ with the same contents."""
+ out = {"reference": reference, "identical": False, "missing": [], "extra":
[], "differing": [], "files": 0}
+ if not os.path.isdir(reference) or not os.path.isdir(other):
+ out["error"] = "not a directory: %s" % (reference if not
os.path.isdir(reference) else other)
+ return out
+
+ def files(root):
+ found = set()
+ for d, _dirs, names in os.walk(root):
+ for n in names:
+ found.add(os.path.relpath(os.path.join(d, n), root))
+ return found
+
+ a, b = files(reference), files(other)
+ out["missing"] = sorted(a - b)
+ out["extra"] = sorted(b - a)
+ for rel in sorted(a & b):
+ if not filecmp.cmp(os.path.join(reference, rel), os.path.join(other,
rel), shallow=False):
+ out["differing"].append(rel)
+ out["files"] = len(a & b)
+ out["identical"] = not (out["missing"] or out["extra"] or out["differing"])
+ return out
+
+
+def resolve_relative_budget(run, results_dir):
+ """Sets the builder budget of a budgetOfPeak build from the reference
build's build.peakBudgetBytes: the latest
+ completed hybrid build of the same dataset with an absolute budget."""
+ ref = None
+ for r in load_results(results_dir): # sorted by run id, so later
timestamps win
+ p = r.get("params") or {}
+ if (r.get("mode") == "build" and p.get("builder") == "hybrid" and
p.get("budgetOfPeak") is None
+ and r.get("datasetName") == run["datasetName"] and
r.get("outcome") == "COMPLETED"
+ and dig(r, "build.peakBudgetBytes")):
+ ref = r
+ if ref is None:
+ raise RuntimeError("budgetOfPeak: no completed unconstrained hybrid
build of %s in %s; run it first"
+ % (run["datasetName"], results_dir))
+ peak = int(dig(ref, "build.peakBudgetBytes"))
+ run["params"]["builderBudget"] = max(1, int(peak *
float(run["params"]["budgetOfPeak"])))
+ run["params"]["budgetReference"] = {"runId": ref["runId"],
"peakBudgetBytes": peak}
+
+
def run_one(run, matrix, classpath, cgm, run_id, results_dir):
e = run["entry"]
+ if run["mode"] == "build" and run["params"].get("budgetOfPeak") is not
None:
+ resolve_relative_budget(run, results_dir)
run_dir = os.path.join(results_dir, run_id)
os.makedirs(run_dir, exist_ok=True)
work_dir = os.path.join(matrix.get("workDir", os.path.join(results_dir,
"work")), run_id)
@@ -760,8 +814,10 @@ def run_one(run, matrix, classpath, cgm, run_id,
results_dir):
except ValueError:
result = None
oom_killed = (exit_code in (-9, 137) and not timed_out) or
bool(cgroup.get("oomKill"))
+ # Spark's uncaught-exception handler exits with 52 (SparkExitCode.OOM) on
an OutOfMemoryError in its threads
+ spark_oom = exit_code == 52 and run["system"] == "spark"
if result is None:
- outcome = "TIMEOUT" if timed_out else ("OOM-KILLED" if oom_killed else
"ERROR")
+ outcome = "TIMEOUT" if timed_out else ("OOM-KILLED" if oom_killed else
("OOM" if spark_oom else "ERROR"))
result = {"mode": run["mode"], "system": run["system"], "dataset":
run["dataset"], "query": run["query"],
"outcome": outcome,
"error": "no result from CsrMemoryRun (exit code %s)" %
exit_code, "heap": None}
@@ -782,6 +838,9 @@ def run_one(run, matrix, classpath, cgm, run_id,
results_dir):
result["params"] = run["params"]
result["datasetName"] = run["datasetName"]
result["cold"] = run["cold"]
+ # the JVM leaves system and query empty for builds and opens; the summary
keys on the matrix's values
+ result["system"] = result.get("system") or run["system"]
+ result["query"] = result.get("query") or run["query"]
result["driver"] = {
"startedAt": datetime.datetime.fromtimestamp(started,
datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"wallMillis": wall_millis, "exitCode": exit_code, "timedOut":
timed_out, "oomKilled": oom_killed,
@@ -790,6 +849,8 @@ def run_one(run, matrix, classpath, cgm, run_id,
results_dir):
"gc": parse_gc_log(os.path.join(run_dir, "gc.log"), wall_millis),
"nmt": nmt,
"phases": phases.events,
}
+ if run["mode"] == "build" and e.get("compareTo") and result.get("outcome")
== "COMPLETED":
+ result["identity"] = compare_trees(e["compareTo"], run["snapshot"])
with open(os.path.join(run_dir, "result.json"), "w") as f:
json.dump(result, f, indent=2)
if not matrix.get("keepWorkDir", False):
@@ -834,6 +895,15 @@ RUN_COLUMNS = [
("cgroupOomKill", "driver.cgroup.oomKill"),
("csrPeakBytes", "csr.peakBytes"), ("csrScratchBytes", "csr.scratchBytes"),
("buildPeakScratchBytes", "build.peakScratchBytes"), ("publishedBytes",
"build.publishedBytes"),
+ ("peakBudgetBytes", "build.peakBudgetBytes"), ("budgetOfPeak",
"params.budgetOfPeak"), ("tag", "params.tag"),
+ ("vertexScanSourceMillis", "buildTimers.vertex-scan_source"),
+ ("vertexScanCallbackMillis", "buildTimers.vertex-scan_callback"),
+ ("edgeScanSourceMillis", "buildTimers.edge-scan_source"),
+ ("edgeScanCallbackMillis", "buildTimers.edge-scan_callback"),
+ ("edgeScanResolveMillis", "buildTimers.edge-scan_resolve"),
+ ("ioReadBytes", "io.read_bytes"), ("ioWriteBytes", "io.write_bytes"),
+ ("ioCancelledWriteBytes", "io.cancelled_write_bytes"), ("ioRchar",
"io.rchar"), ("ioWchar", "io.wchar"),
+ ("identical", "identity.identical"), ("identityReference",
"identity.reference"),
("nmtTotalCommitted", "driver.nmt.totalCommitted"),
]
@@ -852,6 +922,17 @@ def load_results(results_dir):
r.setdefault("runId", name)
if not r.get("datasetName"):
r["datasetName"] = dataset_name(r.get("dataset"), None) if
r.get("dataset") else "none"
+ # build timer names contain dots, which dig() treats as path
separators
+ timers = dig(r, "build.timers") or {}
+ r["buildTimers"] = dict((k.replace(".", "_"), v) for k, v in
timers.items())
+ # results written before the driver filled in system and query for
builds and opens
+ params = r.get("params") or {}
+ if r.get("mode") == "build" and not r.get("system"):
+ r["system"] = params.get("builder")
+ if params.get("builderBudget") is not None:
+ r["query"] = "b%s-%s" %
(human_bytes(params["builderBudget"]), params.get("source"))
+ elif r.get("mode") == "open" and not r.get("query"):
+ r["query"] = "cold" if r.get("cold") else "warm"
out.append(r)
return out
@@ -906,7 +987,8 @@ def summarize(results_dir):
agree.setdefault((ds, q), {})[system] = (dig(r, "result.count"),
dig(r, "result.hash"))
cols = ["mode", "system", "dataset", "query", "outcome", "peakAfterGc",
"peakRss", "peakCgroupMemory",
"anonAtPeak", "fileAtPeak", "coldMillis", "warmMedianMillis",
"scratchBytes", "resultCount",
- "resultHash", "agreement", "agreementDetail", "runId"]
+ "resultHash", "agreement", "agreementDetail", "buildMillis",
"builderBudget", "peakBudgetBytes",
+ "publishedBytes", "ioReadBytes", "ioWriteBytes", "identical",
"runId"]
with open(os.path.join(results_dir, "summary.csv"), "w", newline="") as f:
w = csv.writer(f)
w.writerow(cols)
@@ -939,7 +1021,10 @@ def summarize(results_dir):
w.writerow([fmt(x) for x in [
mode, system, ds, q, r.get("outcome"), after, dig(r,
"driver.os.peakRss"), cg_peak, anon, fil,
dig(r, "timings.coldMillis"), dig(r,
"timings.warmMedianMillis"), scratch,
- dig(r, "result.count"), dig(r, "result.hash"), ag, detail,
r.get("runId")]])
+ dig(r, "result.count"), dig(r, "result.hash"), ag, detail,
dig(r, "timings.buildMillis"),
+ dig(r, "params.builderBudget"), dig(r,
"build.peakBudgetBytes"), dig(r, "build.publishedBytes"),
+ dig(r, "io.read_bytes"), dig(r, "io.write_bytes"), dig(r,
"identity.identical"),
+ r.get("runId")]])
log("wrote %s and %s (%d runs)" % (os.path.join(results_dir, "runs.csv"),
os.path.join(results_dir,
"summary.csv"), len(results)))
@@ -1028,6 +1113,20 @@ def cmd_run(args):
return 0
+def cmd_identical(args):
+ worst = 0
+ for other in args.others:
+ res = compare_trees(args.reference, other)
+ state = "identical" if res["identical"] else "DIFFERENT"
+ log("%s: %s (%d common files)" % (other, state, res["files"]))
+ for k in ("error", "missing", "extra", "differing"):
+ if res.get(k):
+ log(" %s: %s" % (k, res[k]))
+ if not res["identical"]:
+ worst = 1
+ return worst
+
+
def cmd_summarize(args):
if args.reparse:
reparse(args.results_dir)
@@ -1062,6 +1161,11 @@ def main(argv=None):
help="also re-parse gc.log, NMT output and os-series.csv
into result.json")
p.set_defaults(fn=cmd_summarize)
+ p = sub.add_parser("identical", help="compare snapshot directories byte
for byte against a reference")
+ p.add_argument("reference")
+ p.add_argument("others", nargs="+")
+ p.set_defaults(fn=cmd_identical)
+
args = ap.parse_args(argv)
return args.fn(args)
diff --git a/spark-gremlin/src/test/csr-memory/matrix.json
b/spark-gremlin/src/test/csr-memory/matrix.json
index 6ae3b6173e..f975a7a5fe 100644
--- a/spark-gremlin/src/test/csr-memory/matrix.json
+++ b/spark-gremlin/src/test/csr-memory/matrix.json
@@ -2,7 +2,7 @@
"java": "java",
"classpathFile": "/data/csrbench/classpath.txt",
"jvmArgs": [
- "-Xmx48g"
+ "-Xmx96g"
],
"timeoutSeconds": 7200,
"workDir": "/data/csrbench/work",
@@ -18,7 +18,7 @@
"snapshot": "/data/csrbench/snap/powerlaw-16-streaming",
"builder": "streaming",
"source": "gryo",
- "builderBudget": 38654705664
+ "builderBudget": 77309411328
},
{
"mode": "build",
@@ -26,7 +26,7 @@
"snapshot": "/data/csrbench/snap/powerlaw-18-streaming",
"builder": "streaming",
"source": "gryo",
- "builderBudget": 38654705664
+ "builderBudget": 77309411328
},
{
"mode": "build",
@@ -34,7 +34,7 @@
"snapshot": "/data/csrbench/snap/powerlaw-22-ef24-streaming",
"builder": "streaming",
"source": "gryo",
- "builderBudget": 38654705664
+ "builderBudget": 77309411328
},
{
"mode": "build",
@@ -42,7 +42,7 @@
"snapshot": "/data/csrbench/snap/powerlaw-18-heap",
"builder": "heap",
"source": "gryo",
- "builderBudget": 38654705664
+ "builderBudget": 77309411328
},
{
"mode": "open",
@@ -106,7 +106,7 @@
"se-aggregate-count",
"facade-path"
],
- "csrBudget": 38654705664
+ "csrBudget": 77309411328
},
{
"mode": "query",
@@ -121,7 +121,7 @@
"sort-full-count",
"iter-repeat-both-emit"
],
- "csrBudget": 38654705664
+ "csrBudget": 77309411328
},
{
"mode": "query",
@@ -149,7 +149,7 @@
"se-aggregate-count",
"facade-path"
],
- "csrBudget": 38654705664
+ "csrBudget": 77309411328
},
{
"mode": "query",
@@ -164,7 +164,7 @@
"sort-full-count",
"iter-repeat-both-emit"
],
- "csrBudget": 38654705664
+ "csrBudget": 77309411328
},
{
"mode": "query",
@@ -188,7 +188,7 @@
"se-aggregate-count",
"facade-path"
],
- "csrBudget": 38654705664
+ "csrBudget": 77309411328
}
]
}
diff --git
a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
index 9535680e4d..d2f1f0cfb8 100644
---
a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
+++
b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
@@ -35,6 +35,7 @@ import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.snapshot.build.BuildOptions;
import org.apache.tinkerpop.gremlin.structure.snapshot.build.BuildStats;
import
org.apache.tinkerpop.gremlin.structure.snapshot.build.HeapSnapshotBuilder;
+import
org.apache.tinkerpop.gremlin.structure.snapshot.build.HybridSnapshotBuilder;
import org.apache.tinkerpop.gremlin.structure.snapshot.build.SnapshotBuilder;
import
org.apache.tinkerpop.gremlin.structure.snapshot.build.StreamingSnapshotBuilder;
import org.apache.tinkerpop.gremlin.structure.snapshot.graph.CsrGraph;
@@ -85,7 +86,7 @@ import java.util.Set;
* --warm <n> warm runs after the cold run,
default 3
* --timeout-seconds <n> soft timeout for the whole
run, outcome TIMEOUT
* --csr-budget <bytes> csrMemoryBudget of csr-native,
default 1073741824
- * --builder heap|streaming (build) default streaming
+ * --builder heap|streaming|hybrid (build) default streaming
* --source gryo|tinkergraph (build) default gryo
* --builder-budget <bytes> (build) default 268435456
* --verify-checksums true|false (open, csr) default false
@@ -123,7 +124,9 @@ import java.util.Set;
* <p/>
* <b>Build and open.</b> Build mode loads the TinkerGraph first when {@code
--source tinkergraph} (timed separately as
* {@code loadMillis}) and then times the build; {@code build.phases} are the
{@link BuildStats} phases, and
- * {@code peakScratchBytes} is {@link BuildStats#peakDiskBytes()}. Open mode
times {@link CsrGraph#open} (or the Gryo
+ * {@code peakScratchBytes} is {@link BuildStats#peakDiskBytes()}, {@code
build.timers} are the
+ * {@link BuildStats#timers()} in milliseconds (the edge-scan split) and
{@code build.peakBudgetBytes} is
+ * {@link BuildStats#peakBudgetBytes()}. Open mode times {@link CsrGraph#open}
(or the Gryo
* load for tinkergraph), records the heap after a full GC in {@code
extra.retainedHeapBytes}, and then runs
* {@code g.V().count()} and {@code g.V().out().count()} cold ({@code
coldMillis} is the count; the one-hop times are in
* {@code extra}) and once warm.
@@ -153,7 +156,7 @@ public final class CsrMemoryRun {
"Usage: CsrMemoryRun --mode build|open|query [options]",
" --system
tinkergraph|csr-native|csr-facade|tinkergraph-computer|spark (open:
tinkergraph|csr)",
" --dataset <.kryo> --snapshot <dir> --query-file <file>
--query <id> --warm <n> (3)",
- " --timeout-seconds <n> --csr-budget <bytes> (1073741824)
--builder heap|streaming (streaming)",
+ " --timeout-seconds <n> --csr-budget <bytes> (1073741824)
--builder heap|streaming|hybrid (streaming)",
" --source gryo|tinkergraph (gryo) --builder-budget <bytes>
(268435456)",
" --verify-checksums true|false (false) --spark-master <url>
(local[*]) --spark-conf k=v (repeatable)",
" --work-dir <dir> --sample-interval-ms <n> (1000)
--series-out <file> --out <file> --help",
@@ -516,7 +519,8 @@ public final class CsrMemoryRun {
final boolean fromGryo = "gryo".equals(config.source);
if (!fromGryo) loadTinkerGraph();
- final SnapshotBuilder builder = "heap".equals(config.builder) ? new
HeapSnapshotBuilder() : new StreamingSnapshotBuilder();
+ final SnapshotBuilder builder = "heap".equals(config.builder) ? new
HeapSnapshotBuilder()
+ : "hybrid".equals(config.builder) ? new
HybridSnapshotBuilder() : new StreamingSnapshotBuilder();
final BuildOptions options = BuildOptions.builder()
.memoryBudgetBytes(config.builderBudget)
.scratchDirectory(scratchDir("build-scratch"))
@@ -544,6 +548,10 @@ public final class CsrMemoryRun {
buildInfo.put("phases", phases);
buildInfo.put("publishedBytes", stats.totalSegmentBytes());
buildInfo.put("peakScratchBytes", stats.peakDiskBytes());
+ final Map<String, Object> timers = new LinkedHashMap<>();
+ for (final Map.Entry<String, Long> t : stats.timers().entrySet())
timers.put(t.getKey(), t.getValue() / 1_000_000L);
+ buildInfo.put("timers", timers);
+ buildInfo.put("peakBudgetBytes", stats.peakBudgetBytes());
extra.put("builder", config.builder);
extra.put("source", config.source);
extra.put("builderBudget", config.builderBudget);
@@ -659,6 +667,25 @@ public final class CsrMemoryRun {
// ---------------------------------------------------------------- result
+ /**
+ * The process's I/O counters from {@code /proc/self/io}, read when the
result is written so that the last writes
+ * are counted. {@code read_bytes} and {@code write_bytes} are
storage-level (pages read from disk, pages dirtied,
+ * including through memory maps); {@code rchar} and {@code wchar} are
bytes passed to read and write system calls.
+ * Null where the file does not exist.
+ */
+ private static Map<String, Object> readProcIo() {
+ try {
+ final Map<String, Object> io = new LinkedHashMap<>();
+ for (final String line :
Files.readAllLines(Paths.get("/proc/self/io"), StandardCharsets.US_ASCII)) {
+ final int colon = line.indexOf(':');
+ if (colon > 0) io.put(line.substring(0, colon).trim(),
Long.parseLong(line.substring(colon + 1).trim()));
+ }
+ return io;
+ } catch (IOException | RuntimeException e) {
+ return null;
+ }
+ }
+
private synchronized void writeResult() {
if (written) return;
written = true;
@@ -704,6 +731,7 @@ public final class CsrMemoryRun {
result.put("csr", null);
}
result.put("build", buildInfo);
+ result.put("io", readProcIo());
final Map<String, Object> jvm = new LinkedHashMap<>();
jvm.put("version", System.getProperty("java.version"));
@@ -1174,7 +1202,7 @@ public final class CsrMemoryRun {
if (c.system != null && !SYSTEMS.contains(c.system) &&
!c.system.equals("csr")) {
throw new UsageException("unknown system " + c.system, false);
}
- if (!Set.of("heap", "streaming").contains(c.builder)) throw new
UsageException("unknown builder " + c.builder, false);
+ if (!Set.of("heap", "streaming", "hybrid").contains(c.builder))
throw new UsageException("unknown builder " + c.builder, false);
if (!Set.of("gryo", "tinkergraph").contains(c.source)) throw new
UsageException("unknown source " + c.source, false);
return c;
}
diff --git
a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/CsrSnapshotBuildBenchmark.java
b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/CsrSnapshotBuildBenchmark.java
index f7c503bfbc..c955af8f04 100644
---
a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/CsrSnapshotBuildBenchmark.java
+++
b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/CsrSnapshotBuildBenchmark.java
@@ -30,6 +30,7 @@ import
org.apache.tinkerpop.gremlin.structure.snapshot.CsrSnapshot;
import org.apache.tinkerpop.gremlin.structure.snapshot.build.BuildOptions;
import org.apache.tinkerpop.gremlin.structure.snapshot.build.BuildStats;
import
org.apache.tinkerpop.gremlin.structure.snapshot.build.HeapSnapshotBuilder;
+import
org.apache.tinkerpop.gremlin.structure.snapshot.build.HybridSnapshotBuilder;
import org.apache.tinkerpop.gremlin.structure.snapshot.build.SnapshotBuilder;
import
org.apache.tinkerpop.gremlin.structure.snapshot.build.StreamingSnapshotBuilder;
import org.apache.tinkerpop.gremlin.structure.snapshot.format.ColumnReader;
@@ -60,9 +61,9 @@ import java.util.stream.Stream;
* <p/>
* This is a manually invoked spike utility, not a test. It loads the Gryo
file into a {@link TinkerGraph} (or, with
* {@code --source gryo}, streams it through {@link GryoSnapshotSource} and
loads the graph only for {@code --verify}), builds a
- * snapshot with the materializing ({@code heap}) builder, the {@code
streaming} builder or both, and reports phase
- * times, peak on-disk bytes, peak heap and published bytes. {@code --verify}
checks each snapshot against the graph and
- * {@code --compare} checks that the two builders produce identical bundles.
Run it with no arguments for usage.
+ * snapshot with the materializing ({@code heap}) builder, the {@code
streaming} builder, the {@code hybrid} builder or
+ * several of them, and reports phase times, peak on-disk bytes, peak heap and
published bytes. {@code --verify} checks
+ * each snapshot against the graph and {@code --compare} checks that the
bundles of the builders are identical. Run it with no arguments for usage.
*/
public final class CsrSnapshotBuildBenchmark {
@@ -73,11 +74,13 @@ public final class CsrSnapshotBuildBenchmark {
" --source <tinkergraph|gryo> build from the TinkerGraph loaded
from the input or stream the Gryo",
" file directly; gryo loads the
TinkerGraph only for --verify",
" (default tinkergraph)",
- " --builder <heap|streaming|both>",
- " builder(s) to run (default heap;
--compare implies both)",
+ " --builder <heap|streaming|hybrid|both|all|list>",
+ " builder(s) to run: one name, a
comma-separated list such as heap,hybrid,",
+ " both (heap and streaming) or all
(default heap; --compare without --builder",
+ " implies both)",
" --layout <topology|identity|full>",
" layout to publish (default full)",
- " --memory-budget <bytes> streaming builder memory budget; k,
m or g suffix allowed (default 256m)",
+ " --memory-budget <bytes> streaming and hybrid builder memory
budget; k, m or g suffix allowed (default 256m)",
" --grouped-fast-path <true|false>",
" let builders use the
grouped-by-out-vertex path (default true)",
" --edge-id-index <true|false>",
@@ -86,7 +89,8 @@ public final class CsrSnapshotBuildBenchmark {
" contain them (default: a new
directory under java.io.tmpdir)",
" --scratch-dir <dir> scratch directory for the builders
(default: the build directory)",
" --verify check each snapshot against the
TinkerGraph",
- " --compare build with both builders and report
the first differing file");
+ " --compare compare the bundle of the first
builder with each of the others and report the",
+ " first differing file");
private static final int MAX_FAILURES = 20;
private static final long SAMPLE_INTERVAL_MILLIS = 10;
@@ -151,14 +155,18 @@ public final class CsrSnapshotBuildBenchmark {
try (SnapshotSource source = config.source == Source.GRYO
? new GryoSnapshotSource(config.input) : new
TinkerGraphSnapshotSource(graph)) {
for (final String name : config.builders) {
- final SnapshotBuilder builder = "heap".equals(name) ? new
HeapSnapshotBuilder() : new StreamingSnapshotBuilder();
+ final SnapshotBuilder builder = "heap".equals(name) ? new
HeapSnapshotBuilder()
+ : "hybrid".equals(name) ? new HybridSnapshotBuilder()
: new StreamingSnapshotBuilder();
final Path target = outputDir.resolve(name);
ok &= buildAndReport(graph, source, name, builder, target,
options, config);
built.put(name, target);
}
if (config.compare) {
- ok &= compare(built.get("heap"), built.get("streaming"));
+ final String first = config.builders.get(0);
+ for (final String other : config.builders.subList(1,
config.builders.size())) {
+ ok &= compare(first, built.get(first), other,
built.get(other));
+ }
}
} finally {
if (graph != null) graph.close();
@@ -192,7 +200,15 @@ public final class CsrSnapshotBuildBenchmark {
for (final BuildStats.Phase phase : stats.phases()) {
System.out.printf(Locale.ROOT, " %-24s %12s %18s%n",
phase.name(), millis(phase.elapsedNanos()), bytes(phase.diskBytes()));
}
+ if (!stats.timers().isEmpty()) {
+ for (final Map.Entry<String, Long> timer :
stats.timers().entrySet()) {
+ System.out.printf(Locale.ROOT, " %-24s %12s%n", " " +
timer.getKey(), millis(timer.getValue()));
+ }
+ }
System.out.printf(Locale.ROOT, " total build time: %s%n",
millis(elapsed));
+ if (stats.peakBudgetBytes() > 0) {
+ System.out.printf(Locale.ROOT, " peak budgeted heap: %s%n",
bytes(stats.peakBudgetBytes()));
+ }
System.out.printf(Locale.ROOT, " peak on-disk bytes: %s%n",
bytes(stats.peakDiskBytes()));
System.out.printf(Locale.ROOT, " peak heap: %s (used heap
after GC before build: %s, sampled every %d ms)%n",
bytes(sampler.peak.get()), bytes(baseline),
SAMPLE_INTERVAL_MILLIS);
@@ -281,27 +297,28 @@ public final class CsrSnapshotBuildBenchmark {
* Reports the first file, in ascending relative path order, that exists
in only one bundle or differs in content,
* along with the byte offset of the difference.
*/
- private static boolean compare(final Path heap, final Path streaming)
throws IOException {
- System.out.printf("%n== compare heap vs streaming%n");
+ private static boolean compare(final String nameA, final Path first, final
String nameB, final Path second)
+ throws IOException {
+ System.out.printf("%n== compare %s vs %s%n", nameA, nameB);
final TreeSet<String> paths = new TreeSet<>();
- paths.addAll(listFiles(heap));
- paths.addAll(listFiles(streaming));
+ paths.addAll(listFiles(first));
+ paths.addAll(listFiles(second));
for (final String path : paths) {
- final Path a = heap.resolve(path);
- final Path b = streaming.resolve(path);
+ final Path a = first.resolve(path);
+ final Path b = second.resolve(path);
if (!Files.exists(a)) {
- System.out.println(" DIFFERENT: " + path + " exists only in
the streaming bundle");
+ System.out.println(" DIFFERENT: " + path + " exists only in
the " + nameB + " bundle");
return false;
}
if (!Files.exists(b)) {
- System.out.println(" DIFFERENT: " + path + " exists only in
the heap bundle");
+ System.out.println(" DIFFERENT: " + path + " exists only in
the " + nameA + " bundle");
return false;
}
final long offset = firstDifference(a, b);
if (offset >= 0) {
- System.out.printf(Locale.ROOT, " DIFFERENT: %s at byte offset
%,d (heap size %,d, streaming size %,d)%n",
- path, offset, Files.size(a), Files.size(b));
+ System.out.printf(Locale.ROOT, " DIFFERENT: %s at byte offset
%,d (%s size %,d, %s size %,d)%n",
+ path, offset, nameA, Files.size(a), nameB,
Files.size(b));
return false;
}
}
@@ -699,19 +716,7 @@ public final class CsrSnapshotBuildBenchmark {
break;
case "--builder":
builderGiven = true;
- switch (value.toLowerCase(Locale.ROOT)) {
- case "heap":
- config.builders = List.of("heap");
- break;
- case "streaming":
- config.builders = List.of("streaming");
- break;
- case "both":
- config.builders = List.of("heap",
"streaming");
- break;
- default:
- throw new UsageException("Unknown
builder: " + value);
- }
+ config.builders = parseBuilders(value);
break;
case "--layout":
try {
@@ -744,13 +749,40 @@ public final class CsrSnapshotBuildBenchmark {
if (config.input == null) throw new UsageException("--input is
required");
if (!Files.isRegularFile(config.input)) throw new
UsageException("Input file does not exist: " + config.input);
if (config.compare) {
- if (builderGiven && config.builders.size() != 2)
- throw new UsageException("--compare needs both builders;
drop --builder or use --builder both");
- config.builders = List.of("heap", "streaming");
+ if (!builderGiven) config.builders = List.of("heap",
"streaming");
+ if (config.builders.size() < 2)
+ throw new UsageException("--compare needs at least two
builders; drop --builder or use --builder both");
}
return config;
}
+ private static List<String> parseBuilders(final String value) {
+ final List<String> names = new ArrayList<>();
+ for (final String part :
value.toLowerCase(Locale.ROOT).split(",")) {
+ final List<String> expanded;
+ switch (part.trim()) {
+ case "heap":
+ case "streaming":
+ case "hybrid":
+ expanded = List.of(part.trim());
+ break;
+ case "both":
+ expanded = List.of("heap", "streaming");
+ break;
+ case "all":
+ expanded = List.of("heap", "streaming", "hybrid");
+ break;
+ default:
+ throw new UsageException("Unknown builder: " + part);
+ }
+ for (final String name : expanded) {
+ if (names.contains(name)) throw new
UsageException("Builder given twice: " + name);
+ names.add(name);
+ }
+ }
+ return names;
+ }
+
private static boolean parseBoolean(final String option, final String
value) {
if (value.equalsIgnoreCase("true")) return true;
if (value.equalsIgnoreCase("false")) return false;
