perf(index): prevent two-file shuffle starvation

[BubbleCal]

What is the performance issue?

TwoFileShuffleReader resolves offsets and reads data independently for every IVF partition. With many partitions and flush groups this creates millions of small ranges. The ordered partition build stream also stops admitting useful work when an early partition is slow, even if later partitions have already completed.

How does this PR improve performance?

• Validate and sequentially preload the complete offsets table once, with a 256 MiB preload limit and validated on-demand fallback.
• Read contiguous partition windows with a 128 MiB decoded-size target, issuing at most one data range per non-empty flush group, then split the decoded stream back into exact partition order.
• Build fresh partitions with bounded unordered scheduling per index build: 512 MiB decoded-input admission, at most 2 * workers admitted entries, and exclusive admission for an oversized hotspot partition. A shuffle contained in one window can use the complete entry budget; multiple windows retain per-window reservations to prevent ordered-write starvation.
• Admit partition inputs in source order before unordered decoding/building, preventing later partitions from consuming every entry permit while the ordered writer waits for an earlier partition.
• Give the legacy one-file-per-partition reader a conservative schema-based admission estimate instead of forcing every non-empty partition through exclusive admission.
• Carry partition IDs through build results and drain a bounded reorder map strictly in partition order before writing.
• Preserve the existing singleton/ordered behavior for incremental, split, join, and remap paths.

This does not change HNSW graph quality parameters.

Benchmark

The reader benchmark uses the same persistent local files for both implementations and includes a benchmark-local copy of the pre-change reader. It contains 4,096 partitions, 20 flush groups, 262,144 rows, and a 256-dimensional RQ5-like payload, with uniform and CV=3.6001 hotspot distributions.

Environment: Apple arm64, 10 CPU cores / 8 compute workers, 24 GiB RAM, local object store, release-with-debug, cache-hot diagnostic pass. Values are the median of three fresh processes per implementation at PR head 04a64651f. Lower is better for elapsed time, I/O, ranges, and RSS; higher is better for throughput.

Scenario / metric	Baseline	This PR	Benefit
Uniform total elapsed	6,800.322 ms	82.961 ms	81.97x speedup
Uniform throughput	38,549 rows/s	3,159,827 rows/s	81.97x higher
Hotspot total elapsed	6,726.514 ms	50.202 ms	133.99x speedup
Hotspot throughput	38,972 rows/s	5,221,832 rows/s	133.99x higher
Uniform logical data ranges	81,920 ranges	20 ranges	4,096x fewer
Uniform scheduler read IOPS	786,424 operations	22 operations	35,746.55x fewer
Uniform physical bytes read	2,783,291,368 B	47,680,010 B	58.37x fewer
Uniform peak RSS	24.7 MiB	222.7 MiB	9.02x higher (tradeoff)
Hotspot peak RSS	30.7 MiB	222.2 MiB	7.24x higher (tradeoff)

These measurements cover the two-file shuffle read path, not full IVF-HNSW-RQ build throughput. The 128 MiB reader window is materialized, so the speedup intentionally trades additional bounded memory for much lower I/O amplification. The 512 MiB decoded-input admission budget and entry limit apply independently to each index build; they are not a process-global admission budget or a strict bound on graph/output RSS.

Concurrent full-build resource check

At PR head 9c96131d9 (before the single-window and legacy concurrency follow-up), a separate full IVF_RQ check used two independent local datasets, each with 262,144 rows, 4,096 partitions, 256 dimensions, 5-bit RQ, and precomputed centroids. A single fresh process sampled its own RSS and OS thread count while running either one build or two concurrent builds. These are single-run observations under release-with-debug, not baseline-versus-PR performance claims, and have not been remeasured at the current head.

Scenario / metric	Single build	Two concurrent builds	Observation
Total input	262,144 rows	524,288 rows	2.00x work
Wall elapsed	1.916 s	4.002 s	2.09x elapsed
Aggregate throughput	136,817 rows/s	131,009 rows/s	0.96x throughput
Peak process RSS	718.4 MiB	996.8 MiB	1.39x RSS
Peak OS threads	50 threads	65 threads	1.30x threads

The observed process resources remained below 2x in this bounded run, but the implementation does not establish a hard process-global limit: decoded-input and CPU admission objects are instantiated per build. Therefore this PR makes no global-admission or arbitrary-concurrency resource-bound claim.

Testing

• cargo test -p lance-index vector::v3::shuffler::tests --lib --no-fail-fast (21 passed)
• cargo test -p lance bounded_partition_stream --lib --no-fail-fast (11 passed)
• cargo test -p lance index::vector::builder::tests::partition_entry_admission_preserves_input_order -- --exact --nocapture
• cargo test -p lance index::vector::builder::tests::single_partition_window_uses_full_entry_budget -- --exact --nocapture
• cargo test -p lance index::vector::builder::tests::fresh_partition_build_runs_multiple_windows_end_to_end -- --exact --nocapture
• cargo test -p lance-index vector::v3::shuffler::tests::legacy_shuffler_uses_schema_estimate_for_parallel_admission -- --exact --nocapture
• cargo test -p lance index::vector::ivf::v2::tests::test_optimize_with_empty_partition -- --exact --nocapture
• cargo test -p lance 'index::vector::ivf::v2::tests::test_knn::test_dataset_too_small::num_deltas_1_1' -- --exact --nocapture
• cargo test -p lance 'index::vector::ivf::v2::tests::test_knn::test_fewer_than_k_results::num_deltas_1_1' -- --exact --nocapture
• uv run pytest -v -s python/tests/test_dataset.py::test_commit_existing_index_segments_accepts_index_metadata
• cargo fmt --all -- --check
• cargo clippy --all --tests --benches -- -D warnings
• cargo bench --profile release-with-debug -p lance-index --bench two_file_shuffle_read -- --test

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The pre-decode admission bug is fixed, but this revision moves the partition-result bound from individual partitions to whole windows. Keep the window permit shared while emitting completed partition builds incrementally so the configured entry cap continues to bound resident results.

[lance-gatekeeper]

This try_collect retains every PartitionBuildResult in the window until the slowest build completes, before the outer stream can yield even one. A 128 MiB window can span thousands of small partitions, so max_entries = 2 * workers now counts one window while all of its HNSW graph/storage results are resident; this removes the partition-result bound and can drive indexing RSS far beyond the decoded-input budget. Stream completed partition results from each admitted window while sharing its permit, instead of collecting the whole window.

Reproducer

I added a disposable test around the exact map_ok(...).try_flatten() expansion used below: one admitted job (byte and entry cap both 1) returned 8 drop-tracked results. After the first output was yielded, the test expected one resident result and observed 8.

cargo test -p lance index::vector::bounded_partition_stream::tests::expanded_window_results_respect_entry_cap --lib -- --exact

Observed failure: left: 8, right: 1.