Parallel Write performance improvements

docs/release notes/4.1.0/695.performance.md

@@ -0,0 +1,98 @@
+### Fully Asynchronous Parallel Graph Index Writes
+
+**Description**
+`OnDiskParallelGraphIndexWriter` (introduced in #608) parallelizes serialization of
+Level-0 (L0) node records to disk using Java's `AsynchronousFileChannel`. Although it
+already used the async channel API, each write was previously submitted and immediately
+blocked on via `Future.get()` before the next one began — writing a node's ordinal, then
+each feature, then its neighbor list as a sequence of blocking round-trips rather than a
+truly asynchronous pipeline. This PR removes that bottleneck so parallel writes take full
+advantage of async I/O:
+
+- **Fast path** (no pre-written features — the common case): each task now packs its
+  entire ordinal range into a single contiguous `ByteBuffer` and issues one
+  `channel.write()` call for the whole range, instead of one blocking write per node field.
+- **Legacy path** (some features already placed on disk via `writeFeaturesInline()`): each
+  task identifies the contiguous byte spans it still owns, submits every write for its
+  entire range up front, and only then waits on the collected futures — letting the OS
+  schedule the full I/O workload instead of alternating write-then-wait per span.
+- The old per-thread scratch `ByteBuffer`, sized to a single record, forced writes to
+  serialize at the buffer level regardless of channel concurrency. It has been removed;
+  each task now allocates its own range-sized (fast path) or per-region (legacy path)
+  buffer instead.
+
+**Also in this PR:** a `parallelGraphConstruction` boolean was added to
+`ConstructionParameters`, letting BenchYAML / AutoBenchYAML test configs opt into
+`OnDiskParallelGraphIndexWriter` for index construction instead of the default serial
+`OnDiskGraphIndexWriter`, without any code changes.
+
+**Performance**
+Example run writing with NVQ + FUSED_ADC features. Before this change, parallel writes
+were ~4x-8x faster than sequential:
+```
+Sequential write: 6074.18 ms
+Parallel write:   1373.73 ms
+Speedup:          4.42x
+```
+After this change, the same comparison shows ~24x-32x speedups:
+```
+Sequential write: 20147.29 ms
+Parallel write:   627.52 ms
+Speedup:          32.11x
+```
+
+**How to Enable**
+
+*Programmatic API* — no change; `OnDiskParallelGraphIndexWriter.Builder` (available since
+#608) is used the same way as before. The throughput improvement is automatic:
+
+```java
+// Serial (existing) path
+var writer = new OnDiskGraphIndexWriter.Builder(graph, outputPath)
+        .withMapper(ordinalMapper)
+        .with(new InlineVectors(dimension))
+        .build();
+
+// Parallel path
+var writer = new OnDiskParallelGraphIndexWriter.Builder(graph, outputPath)
+        .withMapper(ordinalMapper)
+        .with(new InlineVectors(dimension))
+        // Optional tuning:
+        .withParallelWorkerThreads(0)       // 0 = use available processors
+        .withParallelDirectBuffers(false)   // true = off-heap ByteBuffers
+        .build();
+
+writer.write(featureStateSuppliers);
+writer.close();
+```
+
+To supply a shared, externally-managed executor (e.g. to bound total thread count across
+concurrent builds):
+
+```java
+ExecutorService ioPool = Executors.newFixedThreadPool(16);
+var writer = new OnDiskParallelGraphIndexWriter.Builder(graph, outputPath)
+        .withMapper(ordinalMapper)
+        .with(new InlineVectors(dimension))
+        .withExecutor(ioPool)   // caller is responsible for shutdown
+        .build();
+```
+
+*BenchYAML / AutoBenchYAML* — new in this PR: add the following field under
+`construction` in any index-parameter YAML config file:
+
+```yaml
+construction:
+  parallelGraphConstruction: Yes   # default: No
+```
+
+When set to `Yes`, `Grid` uses `OnDiskParallelGraphIndexWriter` for the on-disk build path.
+The field is optional and defaults to `No` (serial writes) if omitted, so existing config
+files require no changes.
+
+**Notes**
+- The writer produces output in the same on-disk format as `OnDiskGraphIndexWriter`; indexes
+  written with either class are interchangeable and loaded with `OnDiskGraphIndex.load()`.
+  This PR changes only the internal write scheduling, not the on-disk format.
+- Write tasks perform blocking file I/O. For best performance supply an I/O-sized thread pool
+  (thread count ≥ logical cores) rather than a compute-sized pool when using `withExecutor()`.

jvector-base/src/main/java/io/github/jbellis/jvector/graph/disk/ParallelGraphWriter.java

@@ -22,7 +22,6 @@
 import io.github.jbellis.jvector.graph.disk.feature.FeatureId;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousFileChannel;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
@@ -67,11 +66,14 @@ class ParallelGraphWriter implements AutoCloseable {
     private final ExecutorService executor;
     private final boolean ownsExecutor;
     private final ThreadLocal<ImmutableGraphIndex.View> viewPerThread;
-    private final ThreadLocal<ByteBuffer> bufferPerThread;
     private final CopyOnWriteArrayList<ImmutableGraphIndex.View> allViews = new CopyOnWriteArrayList<>();
     private final int recordSize;
     private final Path filePath;
     private final int taskMultiplier;
+    // Passed through to NodeRecordTask so each task allocates the right buffer type.
+    // FUTURE IMPROVEMENT: when the legacy path is removed, each task allocates exactly
+    // one range-sized buffer, making the direct/heap distinction more impactful and cleaner.
+    private final boolean useDirectBuffers;
     private static final AtomicInteger threadCounter = new AtomicInteger(0);
 
     /**
@@ -143,6 +145,7 @@ public ParallelGraphWriter(RandomAccessWriter writer,
         this.graph = graph;
         this.filePath = Objects.requireNonNull(filePath);
         this.taskMultiplier = config.taskMultiplier;
+        this.useDirectBuffers = config.useDirectBuffers;
         if (externalExecutor != null) {
             this.executor = externalExecutor;
             this.ownsExecutor = false;
@@ -163,23 +166,17 @@ public ParallelGraphWriter(RandomAccessWriter writer,
             + Integer.BYTES // neighbor count
             + graph.getDegree(0) * Integer.BYTES; // neighbors + padding
 
-        // Thread-local views for safe neighbor iteration
-        // CopyOnWriteArrayList handles concurrent additions safely
+        // Thread-local views for safe neighbor iteration.
+        // CopyOnWriteArrayList handles concurrent additions safely.
         this.viewPerThread = ThreadLocal.withInitial(() -> {
             var view = graph.getView();
             allViews.add(view);
             return view;
         });
-
-        // Thread-local buffers to avoid allocation overhead
-        // Use BIG_ENDIAN to match Java DataOutput specification
-        final int bufferSize = recordSize;
-        final boolean useDirect = config.useDirectBuffers;
-        this.bufferPerThread = ThreadLocal.withInitial(() -> {
-            ByteBuffer buffer = useDirect ? ByteBuffer.allocateDirect(bufferSize) : ByteBuffer.allocate(bufferSize);
-            buffer.order(java.nio.ByteOrder.BIG_ENDIAN);
-            return buffer;
-        });
+        // FUTURE IMPROVEMENT: the old per-thread scratch buffer (bufferPerThread) is removed.
+        // Each task now allocates its own range-sized buffer in the fast path, or per-region
+        // buffers in the legacy path.  The thread-local was sized to a single record and forced
+        // sub-record writes; the new approach eliminates that bottleneck entirely.
     }
 
     /**
@@ -239,23 +236,19 @@ public void writeL0Records(OrdinalMapper ordinalMapper,
                 final int end = endOrdinal;
 
                 Future<Void> future = executor.submit(() -> {
-                    var view = viewPerThread.get();
-                    var buffer = bufferPerThread.get();
-
                     var task = new NodeRecordTask(
-                            start,                    // Start of range (inclusive)
-                            end,                      // End of range (exclusive)
+                            start,              // range start (inclusive)
+                            end,                // range end (exclusive)
                             ordinalMapper,
                             graph,
-                            view,
+                            viewPerThread.get(),
                             inlineFeatures,
                             featureStateSuppliers,
                             recordSize,
-                            baseOffset,               // Base offset (task calculates per-ordinal offsets)
-                            channel,                  // Async file channel for position-based writes
-                            buffer                    // Thread-local buffer
+                            baseOffset,
+                            channel,
+                            useDirectBuffers    // each task allocates its own buffer(s)
                     );
-
                     return task.call();
                 });
 

jvector-examples/src/main/java/io/github/jbellis/jvector/example/AutoBenchYAML.java

@@ -148,15 +148,16 @@ public static void main(String[] args) throws IOException {
 
                     List<BenchResult> datasetResults = Grid.runAllAndCollectResults(ds,
                             config.construction.useSavedIndexIfExists,
-                            config.construction.outDegree, 
+                            config.construction.isParallelGraphConstruction(),
+                            config.construction.outDegree,
                             config.construction.efConstruction,
-                            config.construction.neighborOverflow, 
+                            config.construction.neighborOverflow,
                             config.construction.addHierarchy,
                             config.construction.refineFinalGraph,
-                            config.construction.getFeatureSets(), 
+                            config.construction.getFeatureSets(),
                             config.construction.getCompressorParameters(),
-                            config.search.getCompressorParameters(), 
-                            config.search.topKOverquery, 
+                            config.search.getCompressorParameters(),
+                            config.search.topKOverquery,
                             config.search.useSearchPruning);
                     results.addAll(datasetResults);
 

jvector-examples/src/main/java/io/github/jbellis/jvector/example/BenchYAML.java

@@ -126,6 +126,7 @@ public static void main(String[] args) throws IOException {
 
             Grid.runAll(ds,
                     config.construction.useSavedIndexIfExists,
+                    config.construction.isParallelGraphConstruction(),
                     config.construction.outDegree,
                     config.construction.efConstruction,
                     config.construction.neighborOverflow,