[dian-lun-lin]

Three commits on the on-disk graph compactor, one per topic: sharded compaction, the output's format version, and the cell-walk optimizations that make large merges practical.

1. Sharded compaction sessions

Cassandra's sharded compaction writes K output sstables per compaction, each over a token range. Running K independent compactions over the same N sources repeats everything that does not depend on the shard, codebook training above all, and loses recall: a hash shard keeps 1/K of each node's same-source edges and nothing replaced them (at K=64 a node kept 0.5 of its 32, and the output was nearly bipartite between the sources).

// liveNodes: the live nodes of ALL the shards together
var compactor = new OnDiskGraphIndexCompactor(sources, liveNodes, VectorSimilarityFunction.COSINE, null);
var session = compactor.sharded();
List<OrdinalMapper> shard0 = session.compactShard(shard0LiveNodes, output0);   // per source, the shard's nodes
List<OrdinalMapper> shard1 = session.compactShard(shard1LiveNodes, output1);

• sharded() opens a session; each compactShard merges one output with that shard's nodes as its live set and returns the shard's ordinal mappers. Shards may become known one at a time.
• The inline and sidecar codebooks and the wide diversity code are trained once, on the live nodes of every shard, by the first shard that needs them, and adopted by the rest (QuantizationCompactionStrategy.adoptCompressor).
• A source with under half its nodes live is joined with itself like another source: one more target of the level-0 cell walk, one more graph search at the upper levels. This restores the same-source edges a shard loses.
• A merge over a slice of the sources can leave their top levels without a live node; the output's hierarchy now ends at the highest level that still has one (previously No live nodes found at maxLevel).

Measured on tessera-2M, two 1M sources (1024-d, cosine, degree 32) into K hash shards, total wall time and recall@10 over all shards, 8 cores:

K	one compaction per shard	one session	rebuild of every shard
4	454 s, 0.781	566 s, 0.825	876 s, 0.836
16	981 s, 0.823	567 s, 0.894	946 s, 0.903
64	1203 s, 0.875	627 s, 0.945	886 s, 0.947

The session's cost is flat in K and its recall within a point of a rebuild. At K=4 the same-source join costs more than the shared training saves; from K=16 it wins on both. ShardExp in benchmarks-jmh is the experiment (see CompactorBenchmark.md).

2. withVersion chooses the output's on-disk format version

The compactor wrote OnDiskGraphIndex.CURRENT_VERSION only. withVersion(int) names the version of the output, any one GraphIndexFormatFactory knows; the header follows the version's format, the footer is written only for footer-based versions (5+), and a version that cannot hold the output is rejected up front (fused PQ needs 6, upper layers need 4+). Source versions are detected from their files and may differ from the output's. A session writes every shard at its compactor's version.

3. Cell-walk optimizations for large merges

The level-0 cell walk and the passes around it, tuned for merges whose store is larger than memory:

• Range sizing from the heap and from measurement. The walk's ranges were bounded by fixed budgets and a pair estimate several times too high, so the store was swept several times more often than needed. Budgets now scale with the heap, and after a source's first range the pairs per node come from what that range measured.
• Record announce-ahead. Source records read in region order are announced to the page cache 64 nodes ahead (POSIX_MADV_WILLNEED on the native reader), instead of one demand fault at a time.
• The hub map by position. Level-1 neighbours are positions in the level, resolved once; the two per-node tables of the whole merge are gone, and cells and region ordinals no longer depend on batch completion order.
• Release when level 0 is done. The hub map, cell tables and every worker's scratch are freed before the upper layers are written.
• Direct I/O for a store larger than memory. When the store does not fit beside the heap, each sweep reads runs of cells into per-thread aligned buffers through a direct-I/O descriptor instead of thrashing the page cache (measured on a 1.5 TB store against 400 GB: a third of the machine in reclaim through the mapping).

The walk's budgets, announce-ahead and direct-read decision have package-private setters for the tests only; each has a default the compactor derives from the machine or the sources.

Tests

TestOnDiskGraphIndexCompactor: direct reads and fixed budgets produce the same merge; sharded sessions over full-precision and fused sources (membership, ordinals, vectors, edge mix, rejection of a non-live node); the empty-top-level merge; withVersion round-trips at every version through the versioned loader and checks the rejections. 27 tests.

[github-actions]

Before you submit for review:

• Does your PR follow guidelines from CONTRIBUTIONS.md?
• Did you summarize what this PR does clearly and concisely?
• Did you include performance data for changes which may be performance impacting?
• Did you include useful docs for any user-facing changes or features?
• Did you include useful javadocs for developer oriented changes, explaining new concepts or key changes?
• Did you rebase your branch onto the latest main for regression testing and PR submission?
• Did you trigger regression testing via Run Bench Main and review results?
• Did you adhere to the code formatting guidelines (TBD)
• Did you group your changes for easy review, providing meaningful descriptions for each commit?
• Did you ensure that all files contain the correct copyright header?
• Did you add documentation for this feature to the release notes directory?

If you did not complete any of these, then please explain below.

[dian-lun-lin]

@pkolaczk This PR addresses the sharding. Could you take a look at the new sharding support API?

[MarkWolters]

Found a couple of issues I'd like to see addressed before merging but overall looks really good!