Experiment/cluster rescore prefetch 20260824 - #724
Draft
jshook wants to merge 43 commits into
Draft
Conversation
Introduce an opt-in control surface a host can install on long-running
jvector operations, without touching any existing call site:
- WorkStage / WorkLimiter / ProgressTracker: the admission + reporting
primitives (throttle down, progress up).
- ProgressLimiter: the single combined facet a host implements.
- LeakyBucketLimiter: a default rate-limited WorkLimiter.
Purely additive. Nothing in the library calls these yet; they establish
the seam that host-driven operations (e.g. compaction) can later be
rewritten against.
Add a host-supplied destination abstraction so compaction output can be
written through a caller-owned channel instead of always allocating its
own file:
- SeekableSink / FileChannelSeekableSink: a minimal seekable byte sink
and its file-backed implementation.
- CompactionDestination / FileCompactionDestination: the compaction
output target, resolvable to a SeekableSink.
Purely additive interface types; no existing code is wired to them on
this branch.
Add ParallelExecutor, a minimal parallel-for abstraction that lets an embedding host supply its own execution strategy -- its own pool, or caller-runs on the calling thread -- instead of jvector reaching for ForkJoinPool.commonPool(): - forkJoin(pool): run on a caller-supplied ForkJoinPool - callerRuns(): run inline, no work escaping to a shared pool - forEachInt / forEach: the parallel-for entry points Additive: a standalone interface with no consumers on this branch. It establishes the execution seam that build, quantization, and compaction paths can be rewritten against later.
Add RuntimeMode, a process-level switch (jvector.mode) separating production from diagnostic runs. Unset is production; diagnostic-only work (e.g. verification walks) sits behind this gate and must be opted into explicitly. Additive: a standalone type with no call sites on this branch. It establishes the gate that diagnostic code paths can consult once wired.
Harden the read path for embedding hosts that manage their own mmap
lifecycle, where a stale offset or a close racing an in-flight read
faults the JVM (SIGSEGV) instead of throwing:
- OnDiskGraphIndex: bound record reads. A node id outside the graph
would become a wild offset into the mapped file (garbage, or a
fault); requireValidNode() rejects it up front, and a stale/corrupt
neighbor block now fails with IllegalStateException instead of
consuming garbage ints.
- ReaderSupplier / SimpleMappedReader: document the close() contract --
the raw-release (immediate unmap -> SIGSEGV) vs coordinated
(liveness handshake -> IllegalStateException) families -- so a host
knows a supplier must not be closed until every vended reader is
quiescent.
Non-breaking: internal bounds checks and documentation only, no
signature or call-path changes. The compactor-side memory safety
(drain-on-unwind, truncate-reused-outputs) stays with the compaction
work, where it lives.
use Objects.requireNonNull Co-authored-by: Ashwin Krishna Kumar <nebulousmagneticwind@outlook.com>
Rework the progress SPI per review: startPhase(WorkStage) is now
ProgressTracker's single abstract method, and onProgress(completed,
total) lives on the returned PhaseScope, making the scope the
capability to report:
- progress for a never-started phase is unrepresentable, and
per-phase implementer state (bars, timers) lives in the scope
instance instead of a map keyed by WorkStage
- concurrent phases of the same stage are distinguished by scope
identity
- lambda ergonomics preserved: PhaseScope.close() defaults to a
no-op, so a progress-only tracker is still one expression:
stage -> (completed, total) -> ...
ProgressLimiter.acquire stays unscoped (throttling is aggregate rate,
not phase identity); the logging combinator now routes phase start,
progress, and completion through one scope wrapper.
The GitHub-applied Objects.requireNonNull suggestion in FileCompactionDestination landed without the java.util.Objects import, breaking compilation. Add the import, and keep the "path" message via the two-arg form so the NPE stays as descriptive as the code it replaced (matching the requireNonNull(sink, "sink") idiom in ProgressLimiter).
Per review discussion: 'Target' reads as a static descriptor a caller could construct up front and pass in, which is exactly the confusion the API is meant to prevent. An OutputReservation is inherently live, single-use, per-run state made against the destination — reserve() a fresh region per compaction, commit(bodyLength) fulfills it, close without commit releases it and discards partial output. The destination stays stateless configuration an embedder builds once. CompactionDestination.open() becomes reserve() to match.
Add ParallelExecutor.over(ExecutorService, int parallelism) so embedders
holding a plain ExecutorService get a one-liner alongside forkJoin(...)
and callerRuns(). Parallel streams cannot be hosted on a generic ES
(inside its workers they silently run on the common pool), so the
adapter chunks on the calling thread and submits, with:
- bounded in-flight window (2x parallelism) on the stream paths and
even range-splitting for forEachInt
- nested use from a body degrading to inline execution instead of
starving a bounded pool into deadlock
- drain-before-unwind: on failure or interrupt, unstarted chunks
skip via a cooperative abort flag and every started chunk is
waited out, so the caller never unwinds beneath a running body.
Future.cancel is deliberately unused: cancel(false) succeeds on a
RUNNING FutureTask and get() then returns while the body still
executes (caught by the new tests before it shipped)
- a ForkJoinPool argument delegates to forkJoin(...), where
whole-pipeline stream decomposition is strictly better
The class javadoc gains a 'Choosing a factory' section spelling out the
relative caveats of the three implementations, including that over(...)
distributes only the body while the source is traversed on the calling
thread.
buildBatches enumerated level-0 nodes via source.getNodes(0), which seeks and reads a 4-byte id at every node's record offset — a full random disk scan of the source graphs. When nothing has warmed the page cache first (notably a full-precision compaction, which has no PQ retrain/pre-encode phase to stream the source), that scan is the first cold access and dominates the run: on cohere-10M 2-UNIFORM disk-cold it was ~22 min of a ~34 min compaction. The compactor already holds liveNodes (a FixedBitSet per source) with exactly the live ordinals, so enumerate from it and skip the scan entirely. Also drops dead nodes up front instead of per-batch. cohere-10M 2-UNIFORM full precision, disk-cold: 2038s -> 658s (3.1x), recall 0.6525 unchanged (identical output — same node set, just enumerated in memory). Upper layers keep getNodes (already in-memory, no disk scan). 12/12 TestOnDiskGraphIndexCompactor pass.
Full-precision compaction descends the hierarchy from the global entry node on every cross-source search, reading a full inline vector (getVectorInto) per hop — a disk fault per hop at >RAM scale. Seeding warm-starts each L0 search from entry points near the query: node u's search of source t starts from the members of u's finished same-source neighbors' merged edges that land in t (near u), skipping the descent. Only where candidates come from changes; diversity, scores, and output are unaffected. Always on for full precision; fused compaction is untouched (it already scores hops via RAM-resident ADC codes, so seeding gains nothing there). The win is read locality — the beam explores the target neighborhood instead of descending from afar. A finished node's merged edges are read back from the output file (which already holds them) via CompactWriter.neighborCountFileOffset + a small per-thread read channel, rather than a heap adjacency mirror. Per-node heap is the done-flag + two ordinal maps (~12 B/node) instead of ~142 with a degree-wide mirror (142 GB -> 12 GB at 1B). The done-flag is published after the record is written, so an output-file read of a flagged node always sees committed data. Includes GraphSearcher.initializeWithSeeds and a CompactorBenchmark fix for FULLPRECISION recall measurement in COMPACT mode (was dereferencing the in-memory base vectors, which COMPACT mode does not load).
…e propagation Process L0 sources in ascending live-size order with a barrier between sources, and have each node search only the sources larger than its own. The reverse direction of every source pair is supplied by propagation instead of a search: when a node's beam finds a match in a larger source, the searcher is offered into the match's bounded reverse-candidate list, and the match's diversity selection later unions those offers with its retained same-source edges. Similarity is symmetric and offers carry exact scores, so propagated candidates score identically to what the target's own search would have produced — only where candidates come from changes; diversity, record format, and the write path are untouched. The larger side of every pair therefore runs no cross-source searches, which is the dominant search population when source sizes are skewed (the LSM-typical merge shape). Upper layers keep full symmetric searches: they are a small fraction of search volume and serve as the query's entry structure, so they stay at maximum quality. Memory cost is one transient reverse-candidate buffer during L0, linear in node count, freed before upper layers are compacted.
…st source A node of the largest source runs no forward cross-source searches, so when it also received no reverse candidates its candidate set is exactly its retained same-source edges — and re-running diversity selection over an already-diversity-selected edge set is a fixed point. Skip selection for such nodes entirely: filter dead neighbors, remap ordinals, and write the record directly, preserving the source adjacency and its order. The record bytes are still fully regenerated: the inline vector is read fresh and per-neighbor fused codes come from the post-retrain code cache (or the retrained compressor in the no-cache fallback), so codebook retraining is unaffected. Neighbor vectors are read only in that no-cache fallback; otherwise the only read is the node's own record. The path fires most under skewed merges, where reverse offers concentrate on hub nodes near the smaller sources' data and leave a meaningful fraction of the largest source untouched.
FusedCompactionStrategy pre-encodes every live node's fused code into a single MappedByteBuffer. That buffer is int-indexed, so it cannot exceed 2 GiB, which capped the cache at floor(Integer.MAX_VALUE / codeSize) nodes. The cap scales inversely with dimension, since codeSize grows with it, and it applies to the total node count across all partitions being compacted rather than to any single index. Past the cap the pass was skipped entirely and compaction fell back to re-encoding per edge. That fallback is not a constant-factor penalty. A fused graph stores each node's neighbours' codes inline, so codes are needed once per edge rather than once per node: the pre-encode path is O(N) parallel encodes, while the fallback is O(N * degree) plus a gather of each neighbour's full float vector. Split the region across several mappings, each holding a whole number of codes so no code straddles a boundary. PreEncodedCodeCache owns the arithmetic and the per-thread views; call sites index by ordinal only. Also fixes a latent int overflow in the refine path, where newOrd * cacheCodeSize was computed as int * int and wrapped at the same boundary. TestPreEncodedCodeCache covers the chunked path directly: production chunks are 1 GiB, so every existing compaction test builds graphs far too small to reach a second chunk, leaving the boundary arithmetic otherwise unexercised.
…d at ~134M nodes ReverseCandidateBuffer backs its per-node candidate pools with flat arrays indexed by ordinal * slots. That product is computed in int, so allocation wraps negative once the merged graph exceeds floor(2^31 / REVERSE_CANDIDATE_SLOTS) total nodes (~134M at 16 slots) and setupCrossLink dies with NegativeArraySizeException. The cap applies to the combined node count of all sources in the merge, so it binds exactly where cross-linking matters most: one large mature segment absorbing small flushes. Split the candidate arrays into fixed-power-of-two chunks of ordinals, mirroring the pre-encode cache chunking; chunk lookup is a shift and mask, and the per-node counts array stays flat since it is one int per ordinal. Chunk size is a constructor parameter so a boundary-crossing test can drive it far below the production 32M ordinals per chunk.
The compactor can assign merged ordinals itself, numbering each source's live nodes in PQ-code order under the retrained codebook (setSimilarityOrdinals; the mapping actually used is exposed via effectiveRemappers). Processing order = ordinal order = write order, so records are written sequentially and similar vectors land in adjacent records - consecutive cross-source searches then revisit the pages the previous search faulted in, and the output's layout carries the same locality to the next compaction and to search time. Reverse-candidate slot blocks are allocated lazily on first offer and freed on consumption, so the buffer tracks not-yet-consumed touched targets rather than node count.
Under similarity ordinals, consecutive nodes are near-twin queries, so they share one anchor search per target source. The anchor's exact-ranked result list serves its followers: each member rescores the list against its own query and skips its search when the triangle inequality certifies that no point outside the list can reach its top-K (memberKth <= ThetaD - delta, in the metric's angular/Euclidean distance). Failing certificates first extend the anchor via search resume in small steps; only then does the member run its own search and become the new anchor. Valid for normalized DOT_PRODUCT, COSINE, and EUCLIDEAN, where the underlying distance is a metric; other similarities take the plain search path. Eliminates ~2/3 of cross-source searches on real workloads with recall preserved by construction.
Above-RAM merges are bound by read latency, not device throughput: best-first search keeps one demand read in flight per thread, and the diversity gather faults up to degree vectors serially per node. ReaderSupplier.willNeed (a posix_fadvise WILLNEED on a dedicated descriptor in MemorySegmentReader) starts asynchronous page-cache fetches without blocking, so a single thread can keep several reads in flight. Two call sites: FrontierPrefetchingView mirrors the searcher's candidate queue and hints its top entries - the search's actual next expansions - and gatherFromSameSource batch-hints every retained neighbor's record before the read loop. Hints are advisory only; results are unchanged.
The resident reverse-candidate buffer holds a slot block for every touched target until consumption; offers onto the largest (last-processed) source accumulate for the whole run - tens of GB at hundreds of millions of nodes, and the structure that breaks memory-bounded compaction of billion-node datasets. Offers now append to small per-band write buffers spilled sequentially to per-band files and are replayed once, through the same dedup and top-slots selection, when the band's group is processed. Bands are keyed by (target source, ordinal band): a source's offers are complete exactly when its own group starts, so bands never mix sources. Peak memory becomes O(band width), independent of node count; offers pass through disk once as sequential I/O.
Compacting one source is the similarity re-sort / dead-node rewrite primitive: no cross-source linking runs, every node takes the retained-only fast path, and the output is the same graph laid out in similarity-ordinal order.
A node now offers itself only to the cross-source neighbors its own diversity selection kept - the reverse-edge rule of single-graph Vamana insertion - instead of to every node its searches surfaced. The offer store, slot semantics, and the target's single deferred selection are unchanged; only the admission rule narrows, cutting offer volume by the candidates-to-kept ratio and shrinking spill traffic and slot-cap pressure by the same factor.
…nment The quantization strategy is created before compactGraphImpl runs, and its CompactionContext snapshots the caller's remappers at that point. When similarity ordinals activate, the compactor swaps in the new mappers - but the strategy kept the stale snapshot, so the fused pre-encode pass placed every code at the caller-proposed ordinal while records, vectors, and adjacency were written at the similarity-assigned ones. Every inline neighbor-code copy then fetched the wrong node's code, ADC scores were noise, and search on the merged index collapsed (recall 0.02 vs 0.77 on dpr-gemma-1m 2/TIERED FUSEDPQ; restored to 0.77 with this fix). The strategy must exist before retrain() but the similarity mappers can only be built after it (they order by the retrained PQ's code prefixes), so the context cannot simply be created later. Instead the compactor re-snapshots its state through a new onRemappersUpdated hook immediately after the swap. The sidecar strategy adopts it too: its writeSidecar emits codes in newToOld order, which had the same misalignment.
Compaction chose its own concurrency: a ForkJoinPool it defaulted to PhysicalCoreExecutor.pool(), an ExecutorCompletionService window it sized itself, Arrays.parallelSort onto the common pool, and a codebook retrain that called ProductQuantization.refine with two more jvector-picked pools. An embedder that had bounded an operation to its own thread budget could not override any of it. Everything now runs through the ParallelExecutor the caller supplies, so how the iteration is distributed is the host's decision: - OnDiskGraphIndexCompactor takes a ParallelExecutor plus a stated parallelism. The ForkJoinPool constructors remain as thin delegates over ParallelExecutor.forkJoin(), so existing callers are unaffected. - taskWindowSize no longer sizes an in-flight window -- the executor owns that -- and survives only to pick task granularity. - CompactionContext carries the executor so strategy fan-out (code pre-encode, sidecar encode) is bound the same way. - CompactionSort replaces Arrays.parallelSort, which takes no pool and always decomposes onto the common pool. - ProductQuantization.refine gains a ParallelExecutor overload; the ForkJoinPool overload delegates through forkJoin(), so index-build decomposition is bit-for-bit unchanged. Dropping the completion-service window means batch results are consumed on the worker that produced them rather than on an orchestrating thread. The L0 record path is safe unlocked: each record goes to its own ordinal's offset via a positional write, the ranges are disjoint by construction, WriteResult already holds a cloned buffer, and doneFlag is an AtomicIntegerArray. The upper-layer path takes a lock, because writeUpperLayerNode appends through the sequential writer and accumulates level-1 feature records in an ArrayList. Draining between L0 source groups is still a barrier, which the cross-link offers depend on: forEach blocks until the whole batch list settles.
compact(Path) always wrote a standalone file at offset 0, so an embedder that keeps graphs inside its own container had to compact to a temp file and copy the body in afterwards. compact(CompactionDestination) reserves an output region from the embedder and writes the graph body directly into it, after whatever header the embedder reserved. CompactWriter already carried a startOffset parameter that nothing ever set to anything but 0; it is now threaded from the reservation, through the writer, and into refinement's read-back of the merged graph. The compactor forces before commit(): the reservation is specified to be fulfilled only once the body is durable, and only the compactor knows when its last write landed. commit() reports the body length measured after onAfterClose has truncated the transient pre-encode cache section away, or it would count that scratch space as part of the graph. A failure leaves the reservation uncommitted, which is the signal the embedder discards the partial output on. compact(Path) delegates via toFile() and so now also deletes its partial file on failure and fsyncs before returning, neither of which it did before.
A compaction runs long enough that a host needs to show it moving, and writes hard enough that a host may need to pace it against foreground traffic. Neither was expressible: the compactor only logged. setProgressLimiter installs the embedder's control surface, defaulting to ProgressLimiter.UNLIMITED, which behaves exactly as if no SPI were installed. CompactionStage names the seven phases -- ordinal assignment, codebook retrain, code pre-encode, base layer, upper layers, refinement, sidecar -- so a host can render progress without knowing the compactor's internals. CompactionContext carries the limiter so strategy phases report on the same terms. Byte admission goes in front of the base-layer and sidecar writes, which are the bulk of the output. Admission is per batch rather than per record: one blocking call amortized over ~128 nodes instead of a limiter round-trip per record. An interrupt while throttled aborts the compaction rather than writing through the limit, and restores the interrupt flag so the embedder's own cancellation still observes it. PhaseScope specifies monotonically non-decreasing progress and is written for a single orchestrating caller, but during a fan-out there is no orchestrating thread: an unguarded addAndGet-then-report lets a worker that counted 200 deliver before the worker that counted 150. Advancing the counter and delivering the report happen under one lock, which keeps reports both ordered and serialized. Phase coverage is asserted per phase rather than by watching threads. An escape from the injected executor bypasses it entirely -- no thread the test observes ever misbehaves, the executor simply never gets asked to do the work -- so the test correlates executor entry with the open phase instead.
Above-RAM merges on the deployed xlink-integration tree are read-latency bound, exactly as 985bfe1 predicted. Measured 2026-08-23 on a 1B-row merge (baselines.ibm_datapile_1b_default, 495 GB host): md0 113,136 reads/s at a 4.00 KB mean request size, 99.1% util CPU 4.4% user, 50.1% iowait compaction byte throughput (15 min): 0.000 MiB/s ~31k-batch merges: 39-46 batches/min, vs 5,387-27,530/min on 2026-08-21 40 of 96 RUNNABLE threads sat in the read path -- 31 in RebufferingInputStream.readFully under FusedPQ$PackedNeighbors.readInto, reached from gatherFromOtherSource -> clusterSearchL0 -> extendAnchor. Nothing here changes default behaviour. Each knob defaults to what the tree already did; the point is that three of them were unreachable and one was invisible, so the regime could not be measured or tuned without a rebuild. jvector.compaction.frontierPrefetch=<int> (default 3, as before) FrontierPrefetchingView's class javadoc has always documented this property. The code never read it -- WIDTH was a hardcoded 3. Now it is honoured, clamped to SHADOW_CAP; 0 disables hinting for a baseline arm. The 3 was "the measured knee" on a cache-resident working set, where a displaced hint is pure waste. Above RAM a hint that lands removes a full device stall, so the knee moves out with the miss rate -- and 3 hints per thread cannot supply the queue depth the device wants. jvector.compaction.batchPrefetchDensity=<long> (default 8, as before) computeBaseBatch warms the batch's own L0 records only when the ordinals form a dense range (span <= 8x batch). Under similarity ordering they often do not, and a declined prefetch was indistinguishable from one that ran. Now tunable (0 disables, negative makes it unconditional) and counted: batchPrefetchIssued / batchPrefetchDeclined. jvector.compaction.crossSourceSeedPrefetch=true (new behaviour, default ON) The only batch hint possible in the cross-source path. gatherFromSameSource hints every candidate before reading them because the candidate list is known up front; gatherFromOtherSource had no hint at all, and the comment in computeBaseBatch says why -- "search reads into other sources are data-dependent and stay demand-faulted". But the SEED set is known before any read: it is the exact entry points the beam expands first. Hinting them asynchronously puts several reads in flight before initializeWithSeeds blocks on the first. Counted as seedHints. This is the one default change; set false for a baseline arm. jvector.disk.adviseRandom=false (default true, as before) MADV_RANDOM is right for search -- a point lookup reads one record and stops -- but it means there is NO FALLBACK: with readahead off, every access the targeted warming misses is exactly one page-sized read that coalesces with nothing. That is the 4.00 KB mean request size above. EXPERIMENTAL and process-wide, so it affects search mappings too; it is an A/B knob, not a production default. The principled fix is per-reader advice, a compaction-time supplier that declines it while search keeps it. Also reordered SHADOW_CAP above WIDTH so the clamp does not depend on compile-time constant folding for its correctness. NOT done here, and worth knowing: 670f558 ("prefetch source graphs into the page cache before bulk phases", on origin/source-pretouch) is still not an ancestor of this branch. Cherry-picking it conflicts in all five files because 985bfe1 later rebuilt the same infrastructure. Its ReaderSupplier.prefetch / OnDiskGraphIndex.prefetchL0Records primitives DID survive into this tree -- what did not is any call site that warms a source before the bulk phases. Adding one is the next experiment, and its "skip when sources exceed MemAvailable" guard needs rethinking first: it self-disables in precisely the above-RAM regime that motivates it. A windowed warm, sized to the batch rather than the file, is the shape the ReaderSupplier.prefetch javadoc already recommends. jvector-base and jvector-native both compile.
Restores the call site 670f558 added and this branch never inherited. Its primitives DID survive -- OnDiskGraphIndex.prefetchL0Records and ReaderSupplier.prefetch are both here -- but nothing warms a source before the phases that sweep it, so every bulk read faults a page at a time against MADV_RANDOM mappings with kernel readahead disabled. The guard is windowed, and deliberately NOT 670f558's rule. That commit skips the pass when the sources exceed MemAvailable, which self-disables in exactly the above-RAM regime that motivates it: on the 1B-row merge measured 2026-08-23 the sources are hundreds of GB against a few hundred GB of cache, so the whole-file test fails and the pass never runs. Here the operator caps pretouched ordinals per source instead, warming as much as the cache can actually hold rather than choosing between everything and nothing. jvector.compaction.sourcePretouchMaxNodes default 0 (OFF) Per-source ordinal cap. 0 disables, -1 warms the whole source. jvector.compaction.sourcePretouchWindowNodes default 1048576 Window size, so transient cache demand stays proportional to the window rather than the file -- the shape ReaderSupplier.prefetch's javadoc already recommends. Off by default because whether it pays depends on the consuming phase's ACCESS ORDER, not just on size, and the two bulk phases differ: CODE_PRE_ENCODE sweeps every live node in order, so a warmed window is still resident when the sweep reaches it. PQ_RETRAIN samples randomly across the source, so on an over-capacity source the head is evicted before it is sampled and the pass is pure cost. Measure per phase before enabling. That distinction is the reason this is a knob rather than a default, and it is the part 670f558's size-only guard could not express. Windows stream one at a time rather than fanning out across sources: prefetchL0Records is synchronous, and overlapping streams on one device turns the sequential access this exists to create back into something the elevator has to sort. Whether per-source parallelism wins on a wide array (this host is 5 NVMe behind md0) is worth measuring, which is why the window is a knob. Accounted to the PQ_RETRAIN phase scope rather than given its own CompactionStage: that enum is the host-visible phase surface -- Cassandra reports it as sai_vector_compaction_phase -- and a new constant would change what hosts see for a pass that is off by default. Best-effort throughout: prefetchL0Records is a no-op on suppliers that cannot warm, a window failure logs and moves on, and nothing here can fail a compaction. TestOnDiskGraphIndexCompactor 19/19 pass.
gatherFromOtherSource's clusterMode branch is the only cross-source path with no
batch hint. The seed hint covers the SEEDED branch; FrontierPrefetchingView hints
the searcher's frontier, not the exact-rescore reads that follow it. Two loops in
clusterMode do hold their candidate list before the first read:
- clusterSearchL0's member-rescore loop, over the unscored suffix of the anchor
list, whose node ids are all known when the loop starts
- extendAnchor's rescore loop, where resume() has already produced the complete
SearchResult
Both are the shape gatherFromSameSource already batch-hints; this branch simply
never had it. Hints are advisory and asynchronous, so issuing them back-to-back
puts several reads in flight before the first getVectorInto blocks.
Measured 2026-08-24 on a 1B-row table during a collapsed 126.9M-ordinal merge: 38
of 94 RUNNABLE compaction threads were in clusterSearchL0 and 1 was in
gatherFromSameSource -- the existing batch hint covers the branch carrying 1 of 39
threads. Two prior arms (crossSourceSeedPrefetch, frontierPrefetch=32) failed to
move that merge because neither reaches this path.
OFF by default so enabling it is the single variable in an A/B. The hint count is
logged unconditionally whenever the cluster path ran, so a zero is visible evidence
the gate is off or the hints are not reaching the device -- silence is what hid the
ReaderSupplier default-no-op for two full test cycles.
Three changes, all needed before the next round of arms can be trusted. 1. jvector.compaction.clusterSearch (default true) gates clusterSearchUsable(). Until now there was no way to switch the cluster path off, so the hypothesis that it drives the above-RAM collapse could be supported but never falsified. The measurement that motivates it, from 183 compactor reports spanning 2026-08-21..24 on a 1B-row table and 1.18 BILLION anchor searches: the certification fast path succeeded on 0.0032% of nodes, and resumes-per-node was exactly 1.98 in every single report. CLUSTER_MARGIN/CLUSTER_EXT_STEP = 32/16 = 2, so every node exhausts the margin, fails to certify, and runs the full fallback search anyway -- having first exact-rescored up to 32 results, each a full vector read. The rate was uniform across healthy and collapsed merges and across merge sizes, so this is deterministic overhead, not a mistuning. 2. The batch-progress line now reports ordinals as well as batches. Ordinals per batch varies from ~125 to 4,096 between merges, so batch rates are not comparable across them; reading them as if they were cost this investigation an 8x error in its headline collapse figure. 3. clusterRescoreReads counts the exact rescores the cluster path performs -- every one a full vector read -- and it is logged with the certification rate whenever the cluster path ran. This makes the cost visible per merge instead of inferred from iostat.
ParallelExecutor.forkJoin let a body failure throw straight out of the parallel stream. A fork/join tree propagates that up through join() as soon as one branch throws, so the call could return while sibling bodies were still running on pool workers. For an embedder that owns the memory those bodies read -- a mapped file it unmaps once the call returns -- that is a use-after-unmap SIGSEGV, not an exception it can catch. ForkJoinParallelExecutor wraps the body: a failure is recorded rather than thrown, later elements short-circuit, the stream always completes normally, and the recorded failure is rethrown after join(). Because the stream completed normally, join() returning means every subtask finished. The interface javadoc now states one drain guarantee that every implementation holds, in place of the old per-factory caveat list. Three tests cover it; reverting to the naive implementation fails two of them.
Three compactor fan-outs handed the ParallelExecutor a Stream of coarse tasks: the similarity-ordinal pass (four quarter-million-node tasks per source), code pre-encode (640 chunks), and the L0/upper batch runner. forEach(Stream) gives an executor no element count, so one that batches stream elements for the fine-grained case cannot split these evenly. With Cassandra's executor batching 32 elements per worker, the ordinal pass ran on ONE thread and pre-encode on half the pool. Measured on the live node the night it was switched on: the ordinal pass took 814s of a 17.8-minute 4M-node merge (76% of the merge; the searches were 15%) and 55 minutes of a 137-minute 16M-node merge. Merge throughput at identical shape went from 1.0-1.5 to 4.4-5.0 min per million nodes. All three now dispatch through forEachInt over the list, which hands the executor the count. That is the contract working as written -- the implementation decides how to distribute, and can only do so when it knows how much there is. No stream dispatch remains in graph/disk.
Dispatching the ordinal pass by index (ab90f90) let the executor split it evenly, but the pass only handed over four tasks per source: a fixed quarter-million-node window on ~1M-node sources. Four tasks can occupy four workers however they are split, and sources run one after another. Measured after ab90f90 on a 40-thread pool: 380s for 6.95M ordinals across 7 sources -- 54s per source, which is 993K nodes on 4 threads at the measured 205us/node to within 6%. The executor was no longer the limit; the caller's granularity was. The window is now derived from the stated parallelism so every source produces several tasks per worker, with a floor that keeps the per-task view open and prefetch call amortized.
…prefetch lead
The above-RAM merge measured on 2026-08-23 (113k reads/s at a 4.00 KB mean
request, 50% iowait, 31 of 40 threads in readInto) is concurrency-bound before
it is device-bound. These are the knob-level changes the I/O cost model in
vector_merge_splat_design.md s8 calls for, ahead of any restructuring. Each has
an off switch so a baseline arm is one property.
jvector.compaction.batchOwnRecordHints=true
computeBaseBatch: a batch whose own-record ordinals are too sparse for one
range prefetch now hints every live record via willNeedL0Record (async)
instead of issuing nothing and demand-faulting one page per record.
Counted as ownRecordHints.
jvector.compaction.outputSyncBytes=64MiB (0 disables)
The L0 write consumer calls FileChannel.force(false) every 64 MiB of
base-layer output, bounding the dirty-page debt so the merge's scattered
source reads do not push its own output pages out on the eviction path.
Counted as outputSyncs.
jvector.disk.prefetchLeadBytes=16MiB (0 disables)
MemorySegmentReader.Supplier.prefetch keeps a WILLNEED lead ahead of its
synchronous 64 KB read cursor, advised in 1 MiB steps, so the pretouch
stream consumes reads already in flight instead of riding the kernel's
128 KB readahead ramp on one thread.
Also logs the base-layer hint counters after L0 ("Base-layer IO hints: ..."):
they were incremented before and never reported.
Tests: MemorySegmentReaderTest covers the lead's clamping at both ends of the
file; TestOnDiskGraphIndexCompactor asserts a declined batch falls back to
per-record hints.
Every stage of a compaction now reports through the host's ProgressLimiter and
is logged and timed, uniformly, at the one point all of them pass through.
StageInstrumentation wraps the installed limiter (or UNLIMITED) in the
compactor: each phase logs "Stage X started", its total when first known, one
line per 10% of progress, and "completed: c/t units in N ms"; time spent
blocked in acquire is accumulated; and each run ends with one line —
"Compaction stage times: A=..ms B=..ms | wall, in-stage, throttle wait
(blocked admissions)" — the where-the-time-goes table, emitted by the merge.
The delegate sees exactly the calls it saw before.
Stages that were not observable, or not stages at all:
SOURCE_PRETOUCH new stage; was accounted inside PQ_RETRAIN. Reports warmed
ordinals per window.
PQ_RETRAIN opened and closed with no report in between. The strategy
and retrainer now take the scope (retrain(vsf, scope)):
sample extraction is the first half, refinement the second.
The sidecar path's retrain ran under no stage; it is now
PQ_RETRAIN too.
SIMILARITY_ORDINALS likewise reported nothing; now reports live nodes
encoded per window against the live total.
FINALIZE new stage around onAfterLevels + writeFooter; the final
force in commit() is logged with its duration.
The host's onProgress is its cancellation checkpoint, so PQ_RETRAIN and
SIMILARITY_ORDINALS — minutes to hours at the sizes measured — become
stoppable where they were not. A stage name is one tag value on the host's
phase metric; nothing host-side needs to change.
Tests: TestStageInstrumentation (delegate transparency, log lines, summary,
close-once, throttle accounting); TestOnDiskGraphIndexCompactor asserts every
stage of a run starts, reports, and reaches its stated total, and that the
ordinal pass reports the live node count.
…inal section
Step 1 of vector_merge_splat_design.md s7: construction emits the index's
base-layer structure as a token stream, an additive section of the index file,
so a later merge can derive its plan by one sequential scan instead of
random-access reads of payload-interleaved records.
Grammar, per ordinal in address order: a NODE prefix (bit 7 set, bit 6 live,
bits 0-5 the node's highest level) then the ordinal as an unsigned varint
delta from the previous NODE; then one NB prefix per base-layer edge followed
by the neighbour as a zigzag varint of (neighbour - node). A dead ordinal is a
NODE with the live bit clear and no NB tokens. Under similarity-ordered
ordinals neighbours sit at nearby ordinals and the deltas are short, which is
what makes the stream small; a raw encoding (prefix + 4-byte int) exists to
measure the delta form against.
Layout: [graph body][section][trailer][footer]. The trailer - section length
and magic - sits immediately before the footer's header copy, so the footer
stays last: a reader that has headerOffset looks 12 bytes before it, one that
does not never touches the section, and every other byte of the index is
where it was. Verified byte-for-byte in
TestOnDiskGraphIndex.testTokenStreamSectionIsAdditive.
Emitters:
AbstractGraphIndexWriter.writeFooter every in-memory-graph writer, so a
host's flush and rebuild paths get the section with no change; it lands
inside the length the host measures from the writer's position.
OnDiskGraphIndexCompactor.emitTokenStream stage TOKEN_STREAM, at the end of
the run: refinement rewrites base-layer adjacency after the footer is
written and the pre-encode cache is mapped past the projected end until
onAfterClose, so the finished output is the only correct source. One
sequential read-back of the base layer in 1M-node prefetch windows, then
the footer rewritten behind the section.
Reader: OnDiskGraphIndex.tokenStreamSection() / openTokenStream(), discovered
in loadFromFooter; files without a section load as before.
Knobs: jvector.tokenStream (default true; Builder.withTokenStream),
jvector.tokenStream.encoding (delta | raw), jvector.compaction.tokenStream
(default true). Each emitter logs nodes, edges, bytes, B/edge and the
raw-equivalent size.
Measured on the unit-test graphs (random vectors, not similarity-ordered):
2.3-3.0 B/edge delta against 5.0 raw.
Tests: TestNodeTokenStream (both encodings round-trip, strict locator, strict
encoder); the writer parity test above; the sequential writer with and without
a hierarchy; the compactor with refinement off and on.
Step 2 of vector_merge_splat_design.md s7: the merge derives its plan from the sources' token streams instead of reading every source vector. The plan is a sort of live nodes on a per-node key, and the key in use was the leading bytes of each vector's PQ code under the codebook retrained for that merge - a value that does not exist until merge time and so can never be carried by a source's stream. SimilarityKey replaces it with one every index computes identically at construction, from the vector alone: the sign bits of 32 Gaussian random projections drawn from a fixed seed. Near vectors agree on a hyperplane with probability 1 - theta/pi, so they share their leading bits and sort together, the leading bits dominating the order the way the leading PQ subspaces did. Streams of every generation compare. Token stream v2 carries the key as a 4-byte field after each NODE's ordinal when the header names a key function; v1 streams (no keys) still load. Writers compute it from the inline-vector supplier they already receive at write() - both host paths pass one; NVQ-only or separated-vector graphs get no keys. The compactor computes it from the output's vectors during the read-back it already does. The ordinal pass, per source: if the stream carries this key function, one sequential decode of a few bytes per node, no record touched, no encoding; otherwise the previous windowed vector sweep computing the same key. The default key is therefore lsh; jvector.compaction.similarityKey=pq restores the previous key (vector sweep only) for an A/B. The pass logs "keys from stream for k of S sources". Batches are the plan's per-source windows read off by one forward scan of newToSrc/newToOld (planWindow), replacing the per-source sort of (new, old) pairs. What changed on purpose: the ordinal order itself differs from the PQ-prefix order. Locality under lsh against pq is a measurement for a real merge, not something the unit tests decide. Tests: the plan from stream keys equals the plan from vectors exactly, dead nodes included (testPlanFromStreamMatchesPlanFromVectors); planWindow equals the sort-based construction; keys in every written stream equal the key of the stored vector (assertTokenStreamMatches); v2 round-trips with keys in both encodings; a hand-written v1 section decodes; SimilarityKey is deterministic across instances and negation flips every bit; the similarity-ordinals recall test passes under the new key.
Step 3 of vector_merge_splat_design.md s7. Before a source's base-layer batches run, the source is swept once in its own ordinal order and every live node's vector and base-layer edges are spilled into bands of the merged ordinal space; the batches then read each node's own record from its band instead of seeking into the source. Where this departs from the design's s5: similarity ordinals are assigned source by source, so one source's live nodes occupy one contiguous range of the merged space and its bands are its own. The distribute therefore runs per source, inside the existing source loop - scratch is one source's records, released when its batches finish, not a payload-sized area - and the inter-source reverse-offer barrier is untouched. The pre-encode fold-in is not done: the code cache must be complete before the first source's batches run, which a per-source distribute cannot provide. BandStore: records are appended under a per-band lock to the band of the node's new ordinal (jvector.compaction.bandNodes, default 256K, clamped so a band file stays under 1 GiB), so distribute windows run across the pool. Record: int old | float[dim] | int count | int[degree] - the fused codes are not copied, the output's come from the pre-encode cache, so the spill is about 40% of a fused record. A slot index (new - start -> slot) is kept while distributing; bands are memory-mapped lazily for reads, unmapped and deleted on close. Consumers: the three places that read a node's own record - processBaseNode's vector, gatherFromSameSource's adjacency, the retained-only path - take it from the band when the node's source is the one being processed; the batch's own-record prefetch and hints are skipped, since they would warm pages no longer read. Same-source neighbours' vectors, cross-source searches and diversity vectors still read the sources. Stage DISTRIBUTE, one phase per source, nested inside BASE_LAYER. Logs per source: records, bytes, bands, bytes/record, wall, MB/s; on release: bands mapped, vectors and edge lists served, scratch freed. The base-layer IO line gains "bands: enabled= ownRecordsFromBands= spilled=". jvector.compaction.bandStaged (default true; requires similarity ordinals) disables it. Tests: testBandStagedOutputMatchesDirect - with a single-threaded executor the band-staged output is byte-identical to the direct path's, every live node's own record came from its band, and no spill remains. TestBandStore covers concurrent distribute, dead nodes, multi-band lookup, close-deletes, and the file-size clamp.
testBandStagedOutputMatchesDirect asserted byte equality between the band-staged and direct outputs. It is seed-dependent: under randomized seed 529FE7ED801CAAD3 the files differ inside the header - the retrained PQ codebook's centroid floats - and two runs of the direct path under the same seed differ there too. The codebook refinement is not byte-deterministic (float summation order); nothing the band path touches is affected by it, and every decoded quantity was identical on both arms in every run. The test now compares what the band path can influence: node count, every node's base-layer edges, similarity key and stored vector, and the upper-layer edges, plus the spill-removed check. Passes under the failing seed and a fresh one.
…all experiment Step 4 of vector_merge_splat_design.md s7: candidate scoring from PQ codes instead of candidate vector reads, as a selectable mode, with the recall experiment that decides whether it becomes the default. AdcScorer (quantization): asymmetric distance computation against a code the caller supplies. setQuery builds the per-query partial-sum table over every (subspace, centroid) once; similarityTo(code) is M table lookups. Same formulas as the PQVectors decoders - verified equal to precomputedScoreFunctionFor for dot product, Euclidean and cosine - but over a code handed in, so the merge can score from its pre-encode cache, which holds every live node's code under the retrained codebook keyed by output ordinal. jvector.compaction.candidateScoring=adc (default exact). With it no candidate record is read at all: same-source neighbours are scored from their cached codes against the node's exact vector; cross-source results are rescored from their codes instead of their vectors; the diversity pass compares candidates decoded from their codes. The node's own vector stays exact. The cluster-search path is disabled under adc - it rescores against the anchor's query and certifies on exact distances - so every cross-source candidate comes from the cold search and is code-scored. Requires fused PQ (the cache); without it the mode logs and scores exactly. Counted as codeScores and decodes on the base-layer log. Recall on the unit-test sources (3 x 256 random 32-dim vectors, PQ 8x256, recall@10 over 20 queries, one seed): exact 1.000, adc 0.965, with 30,718 code scores and 35,158 decodes standing in for that many record reads. Random uniform vectors are the least quantizer-friendly data there is; the number that decides the default is the same experiment on real data. The default stays exact. Tests: TestAdcScorer (decoder parity, decode restores the centroid); testAdcScoringRecallAgainstExact runs both modes on the same sources and queries, checks the code path was taken, and guards recall at 0.15 below exact.
Step 5 of vector_merge_splat_design.md s7. Barrier. WritebackAdvisor (base) is the host-interface durability_barrier at band granularity; SyncFileRangeAdvisor (jvector-native, FFM) implements it with sync_file_range(SYNC_FILE_RANGE_WRITE), which queues writeback of exactly the given range and returns, found by reflection the way MemorySegmentReader is. In the base-layer write consumer the batch that completes a band - per-band batch totals from the plan; a batch straddling two bands counts for both - hints the band's output range. With the advisor present the step-0 whole-file fdatasync cadence yields to it: it was blocking and flushed everything. jvector.compaction.bandWriteback. Pretouch. KeyBlockIndex per keyed source - the unsigned min/max SimilarityKey of each run of keyBlockNodes (4096) ordinals - is built during the ordinal pass from the same stream decode that supplies the plan's keys; band key ranges fall out of the sorted plan. The first batch of each band claims it and warms, in every larger source, the blocks whose extent overlaps the band's key range, consecutive blocks merged into runs, capped at bandPretouchMaxNodes (1M) per target. A target where more than bandPretouchMaxOverlap (0.5) of the blocks overlap is arrival-ordered, not key-clustered, and is skipped as unclustered rather than warmed whole. This is the design's s5.7 window argument in code: first-generation sources cost nothing and gain nothing; a merge output, piecewise key-sorted, is a window per band. Measured on the unit-test sources: first generation 3 bands, 3 targets skipped as unclustered, 0 warmed, 3 writeback hints (native); second generation - a merge output re-merged with a fresh source, 64-node bands and blocks - 16 bands, 960 target ordinals warmed (~240 of 768 per searching band), 1 skipped. Logged after L0: "Band barriers: writebackHints= (advisor=) ; band pretouch: bands= nodes= skippedUnclustered=". Tests: TestKeyBlockIndex; SyncFileRangeAdvisorTest (native links, the base factory finds it, hints a written range); testBandBarrierAndPretouch across both generations.
…cency Step 6 of vector_merge_splat_design.md s7: the cross-source beam search walks a source's base-layer adjacency in memory, built from its token stream, and scores every visited node from the pre-encode cache - the search reads no record. ResidentGraph: one source's base-layer adjacency from one sequential stream decode, degree + 1 ints per node (count, then edges, -1 padded) in 256 MB chunks. Its View serves level 0 from the arrays and delegates everything else - upper-level edges, already resident in OnDiskGraphIndex, the entry node, liveness, containment - to the source's own view. GraphSearcher.Builder takes a View, so that is the whole integration surface. jvector.compaction.residentSearch (default off). Sources go resident largest-first within residentGraphMaxBytes (8 GiB) - the largest sources are the search targets - and each thread holds a searcher per resident source. A search over one scores each visited node with AdcScorer against its cached code under the retrained codebook; results are then rescored per candidateScoring: exact is one record read per result rather than per hop, adc is none. The cluster path is off for resident sources. Requires fused PQ and sources that carry a token stream; any other source is searched on disk as before. Logged after the layers: searches, resident sources, adjacency bytes. Parity: the resident adjacency equals the on-disk base layer at every level, and a search over the resident view with the same score function as the on-disk view returns identical results - node for node, score for score, same visited count - for 20 queries (TestResidentGraph). Recall on the unit-test sources (recall@10, 20 queries): on-disk 0.995, resident 0.995, 768 searches over 3 resident sources. With residentSearch=true and candidateScoring=adc the base layer's reads are the distribute sweep and cache lookups, nothing else. Default off, like adc: the 2 x 2 experiment on real data decides.
…, mapped read-back The experiment steps 4 and 6 deferred to real data: residentSearch x candidateScoring on the same four flush sstables (3.97M nodes, dim 384, fused PQ 96x256), recall against brute-force ground truth. Results and the setup are in vector_merge_splat_design.md s8.8: recall@10 exact 0.9520, adc 0.9475, resident 0.9485, both 0.9485 - half a point where the synthetic vectors showed 3.5-5.5. MergeExperiment (jvector-examples): retrofit copies an SAI Terms component, drops the container trailer and appends the token stream; gt samples queries from the sources and brute-forces top-K over all of them in one pass; arm merges with the two switches set and reports recall, wall, and the process's own block IO (/proc/self/io), so the daemon's traffic on the same array does not count. TokenStreamRetrofit: the compactor's end-of-run emit, factored out so it can be applied to an existing index - what turns a first-generation source into one the plan, the resident search and the key-window pretouch can use. The compactor delegates to it. The read-back now goes through the mapped reader: the RandomAccessFile-backed one turned every neighbour list into ~35 read(2) calls, 25 us a node, 98-115 s per 4M-node arm; a 1M-node retrofit went from 31.7 s to 5.8 s. OnDiskGraphIndex.discoverTokenStream finds the section for a graph loaded by header offset inside a container that has its own trailer after the graph's footer (a raw SAI component). Public setters for the three per-merge switches: setCandidateScoringFromCodes, setResidentSearch, setBandStaged.
Contributor
|
Before you submit for review:
If you did not complete any of these, then please explain below. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
For experimental review only at this time.
This includes the SPLAT IO scheduling approach and is in active testing.