perf(fts): search partitions in pipelined chunks#7950
Conversation
📝 WalkthroughWalkthroughBM25 inverted-index searches now process partitions in configurable chunks. Postings load concurrently within each chunk, while scoring runs in one CPU task. Tests cover global scoring across impact and legacy partitions. ChangesChunked FTS search
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant InvertedIndex
participant AsyncLoader
participant CpuScorer
InvertedIndex->>AsyncLoader: Load postings for each partition in a chunk
AsyncLoader-->>InvertedIndex: Return non-empty partition data
InvertedIndex->>CpuScorer: Score loaded partitions sequentially
CpuScorer-->>InvertedIndex: Return PartitionCandidates
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
One tiny cpu-pool task per partition (~350 per query) interleaves so heavily under concurrent load that scoring loses cache locality and the dispatch rate floods the blocking pool. Partitions are now searched in chunks of LANCE_FTS_SEARCH_CHUNK (default 16): each chunk loads its postings and scoring DocSets concurrently, then a single cpu task scores the chunk's partitions back-to-back on one thread; chunks pipeline against each other through buffer_unordered. On a 320-core node against a 100M-doc 42-language corpus (5-term OR, k=100, warm), per-query scoring CPU drops ~5x and c16 throughput doubles (227 -> 428 qps at half the latency); combined with a contention-free cache read path the chunked layout reaches 1340 qps at c128 versus 345 for per-partition tasks.
01ca66d to
34e98d5
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance-index/src/scalar/inverted/index.rs`:
- Around line 1000-1004: Replace the unbounded try_join_all fan-out in the chunk
loading path with a bounded unordered stream using self.store.io_parallelism()
as the concurrency limit, while preserving error propagation and flattening all
loaded posting lists into the existing loaded collection.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Pro Plus
Run ID: 60bc4dba-548e-4233-bf90-c85d39ea88f3
📒 Files selected for processing (1)
rust/lance-index/src/scalar/inverted/index.rs
|
Good catch on the arithmetic — the chunked layout does raise the number of pending load futures. We benchmarked the concern directly and added a bound; data below (320-core node, 100M-doc 42-language index, ~350 partitions/query, contention-free cache backend so cache effects don't mask load effects). 1. Cold path (empty cache, fresh process) — chunking helps, doesn't flood:
The reason pending futures don't translate into storage pressure: actual in-flight IO is bounded by the store's scheduler ( 2. Chunk-concurrency bound sweep (warm, the bound you suggested):
The 32-vs-unbounded row is identical by construction, so its 11% delta bounds run-to-run noise at ~±6% — every difference in the table is inside that band. The bound is effectively free, so the code now uses Chunk size itself was swept too (4/8/16/32 at c16 and c128): 16 is the knee (4 costs 24% of peak throughput, 32 is flat vs 16), hence the default. |
a5678ca to
84d8fe5
Compare
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rust/lance-index/src/scalar/inverted/index.rs (1)
105-113: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winReject invalid
LANCE_FTS_SEARCH_CHUNKvalues instead of silently defaulting.Malformed values and
0are silently replaced with16, hiding configuration mistakes and making the effective concurrency differ from the requested value. Keep the default only for an unset variable; reject present invalid values withError::invalid_inputthat includes the variable name and raw value, then propagate the error frombm25_search.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance-index/src/scalar/inverted/index.rs` around lines 105 - 113, Update fts_search_chunk to return a Result and preserve 16 only when LANCE_FTS_SEARCH_CHUNK is unset; for malformed or zero values, return Error::invalid_input including the variable name and raw value. Propagate this Result through bm25_search and update its callers as needed.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@rust/lance-index/src/scalar/inverted/index.rs`:
- Around line 105-113: Update fts_search_chunk to return a Result and preserve
16 only when LANCE_FTS_SEARCH_CHUNK is unset; for malformed or zero values,
return Error::invalid_input including the variable name and raw value. Propagate
this Result through bm25_search and update its callers as needed.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Pro Plus
Run ID: 8147fe44-0f0f-47c8-9793-6fee2b491e63
📒 Files selected for processing (1)
rust/lance-index/src/scalar/inverted/index.rs
Xuanwo
left a comment
Xuanwo
left a comment
There was a problem hiding this comment.
Chunked dispatch addresses a real performance problem, but I found one exactness blocker in the legacy/no-impact path. The reproducer and candidate fixes are attached inline.
| let partition_threshold = if use_global_scorer { | ||
| impact_shared_threshold | ||
| } else { | ||
| legacy_shared_threshold |
There was a problem hiding this comment.
Sharing legacy_shared_threshold across the partitions searched sequentially in a chunk can drop the true global top-k result for legacy/no-impact indexes.
Those partitions use IndexBM25Scorer, whose IDF and document normalization are partition-local, but publish their local kth score into the same threshold. The next partition may therefore prune against a score from a different scale before the surviving candidates are rescored globally. With the reproducer below, the correct global winner is row 200 with score 2.0739653, but the default chunked path returns row 30000 with score 1.3931961.
This is an existing schedule-sensitive threshold bug that the new sequential chunk loop makes stable: on this fixture, the current head with the default chunk failed 20/20 runs, while the exact base and the current head with LANCE_FTS_SEARCH_CHUNK=1 passed 20/20. The latter is therefore a mitigation, not a root fix.
A minimal fix could preserve the batching work while giving each legacy partition a private threshold, reserving the shared threshold for paths that use the globally comparable scorer. Alternatively, legacy pruning could use the global scorer and bounds if those bounds can be made sound. In either case, it would be useful to keep this as a two-legacy-partition regression that asserts both the row id and global score.
If the broader scheduling shape is revisited later, a query-global bounded prepare stage followed by greedy ready micro-batches could retain the dispatch reduction while avoiding the static chunk load barrier. That is a separate tradeoff and is not required to close this correctness issue.
Reproducer run on 8282923f
Add this test next to the existing global-scoring tests:
#[tokio::test]
async fn test_chunked_legacy_threshold_keeps_global_winner() {
let tmpdir = TempObjDir::default();
let store = Arc::new(LanceIndexStore::new(
ObjectStore::local().into(),
tmpdir.clone(),
Arc::new(LanceCache::no_cache()),
));
let first_num_docs = 20_001_u32;
let mut first = InnerBuilder::new_with_format_version(
0,
false,
TokenSetFormat::default(),
InvertedListFormatVersion::V1,
);
first.tokens.add("alpha".to_owned());
first.tokens.add("beta".to_owned());
first
.posting_lists
.push(PostingListBuilder::new_with_posting_tail_codec(
false,
PostingTailCodec::Fixed32,
));
first
.posting_lists
.push(PostingListBuilder::new_with_posting_tail_codec(
false,
PostingTailCodec::Fixed32,
));
for doc_id in 0..first_num_docs {
if doc_id.is_multiple_of(2) {
first.posting_lists[0].add(doc_id, PositionRecorder::Count(1));
}
if !doc_id.is_multiple_of(2) || doc_id + 1 == first_num_docs {
first.posting_lists[1].add(doc_id, PositionRecorder::Count(1));
}
first.docs.append(10_000 + u64::from(doc_id), 5_000);
}
write_test_partition_with_optional_impacts(
&store,
0,
first,
TokenSetFormat::default(),
false,
)
.await;
let mut second = InnerBuilder::new_with_format_version(
1,
false,
TokenSetFormat::default(),
InvertedListFormatVersion::V1,
);
second.tokens.add("alpha".to_owned());
second.tokens.add("beta".to_owned());
second
.posting_lists
.push(PostingListBuilder::new_with_posting_tail_codec(
false,
PostingTailCodec::Fixed32,
));
second
.posting_lists
.push(PostingListBuilder::new_with_posting_tail_codec(
false,
PostingTailCodec::Fixed32,
));
second.posting_lists[0].add(0, PositionRecorder::Count(1));
second.posting_lists[1].add(0, PositionRecorder::Count(1));
second.docs.append(200, 1_000);
for row_id in 201..301 {
second.docs.append(row_id, 1);
}
write_test_partition_with_optional_impacts(
&store,
1,
second,
TokenSetFormat::default(),
false,
)
.await;
write_test_metadata(&store, vec![0, 1], InvertedIndexParams::default()).await;
let cache = LanceCache::with_capacity(64 * 1024 * 1024);
let index = InvertedIndex::load(store, None, &cache).await.unwrap();
index
.prewarm_with_options(&FtsPrewarmOptions::default())
.await
.unwrap();
let tokens = Arc::new(Tokens::new(
vec!["alpha".to_owned(), "beta".to_owned()],
DocType::Text,
));
let params = Arc::new(FtsSearchParams::new().with_limit(Some(1)));
let global = index
.bm25_base_scorer(tokens.as_ref(), params.as_ref())
.await
.unwrap();
let first_partition = index
.partitions
.iter()
.find(|partition| partition.id() == 0)
.unwrap();
let second_partition = index
.partitions
.iter()
.find(|partition| partition.id() == 1)
.unwrap();
let first_local = IndexBM25Scorer::new(std::iter::once(first_partition.as_ref()));
let second_local = IndexBM25Scorer::new(std::iter::once(second_partition.as_ref()));
let first_local_score = ["alpha", "beta"]
.into_iter()
.map(|token| first_local.query_weight(token) * first_local.doc_weight(1, 5_000))
.sum::<f32>();
let second_local_score = ["alpha", "beta"]
.into_iter()
.map(|token| second_local.query_weight(token) * second_local.doc_weight(1, 1_000))
.sum::<f32>();
let first_global_score = ["alpha", "beta"]
.into_iter()
.map(|token| global.query_weight(token) * global.doc_weight(1, 5_000))
.sum::<f32>();
let second_global_score = ["alpha", "beta"]
.into_iter()
.map(|token| global.query_weight(token) * global.doc_weight(1, 1_000))
.sum::<f32>();
assert!(first_local_score > second_local_score);
assert!(second_global_score > first_global_score);
let (row_ids, scores) = index
.bm25_search(
tokens,
params,
Operator::And,
Arc::new(NoFilter),
Arc::new(NoOpMetricsCollector),
None,
)
.await
.unwrap();
assert_eq!(
row_ids,
vec![200],
"scores={scores:?}, first_local={first_local_score}, \
second_local={second_local_score}, first_global={first_global_score}, \
second_global={second_global_score}"
);
}Run:
cargo test -p lance-index \
scalar::inverted::index::tests::test_chunked_legacy_threshold_keeps_global_winner \
-- --exact --nocapture
Current-head failure:
left: [30000]
right: [200]
scores=[1.3931961]
first_local=1.3861945
second_local=0.2211894
first_global=1.3931961
second_global=2.0739653
There was a problem hiding this comment.
Great catch — fixed in b537edf with the minimal approach you suggested: the shared floor is now reserved for the impact (globally comparable) scorer, and each legacy partition seeds a private -inf floor, so cross-partition pruning never sees a score from a different scale. Added the two-legacy-partition regression you asked for (test_two_legacy_partitions_keep_private_thresholds): both partitions in one chunk, asserting the global winner's row id (200) and its globally-rescored BM25 score. Verified it bites — reintroducing the shared legacy floor makes it fail exactly as your reproducer does (returns partition 0's row 100).
Legacy (no-impact) partitions score with partition-local BM25 statistics, so their scores are not comparable across partitions. Publishing them into a threshold shared by the whole query lets one partition's local k-th score prune another partition's true global winner before global rescoring. The sequential chunk loop made this pre-existing schedule-sensitive bug deterministic. Keep the shared floor only for the globally-comparable impact scorer and seed each legacy partition with a private floor. Regression: two legacy partitions in one chunk, asserting the global winner's row id and score.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rust/lance-index/src/scalar/inverted/index.rs (1)
105-111: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winClamp zero to one instead of reverting to the default.
LANCE_FTS_SEARCH_CHUNK=0is filtered out and becomes16, not1; this violates the stated “clamped to at least one” behavior and can unexpectedly increase concurrent work. Usemap(|n| n.max(1)), and add parser-level tests for unset, zero, and a positive value.Proposed fix
- .filter(|&n| n >= 1) + .map(|n: usize| n.max(1)) .unwrap_or(16)As per coding guidelines, “Every bugfix and feature must have corresponding tests.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance-index/src/scalar/inverted/index.rs` around lines 105 - 111, Update fts_search_chunk so parsed values are clamped with n.max(1), causing zero to resolve to one while preserving the default of 16 only when the variable is unset or unparsable. Add parser-level tests covering an unset variable, zero, and a positive value.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@rust/lance-index/src/scalar/inverted/index.rs`:
- Around line 105-111: Update fts_search_chunk so parsed values are clamped with
n.max(1), causing zero to resolve to one while preserving the default of 16 only
when the variable is unset or unparsable. Add parser-level tests covering an
unset variable, zero, and a positive value.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Pro Plus
Run ID: d028f762-f346-4a03-af98-820593194e1a
📒 Files selected for processing (1)
rust/lance-index/src/scalar/inverted/index.rs
) ## Problem moka records every cache **hit** into a single global read-op channel (input for its LRU/TinyLFU policy), and readers trigger inline housekeeping when the channel passes its flush point. Each warm hit therefore touches globally-shared cache lines (`record_read_op`: channel length check + housekeeper try-lock CAS + try_send). At index-cache read rates this collapses: Pure `get` microbenchmark, 50K entries x 4KB, 100% hits, 320-core host: | readers | moka | quick_cache | |---|---|---| | 1 | 2.87 Mops/s (348 ns/op) | 12.05 Mops/s (83 ns/op) | | 64 | 0.74 (1345 ns) | 11.03 (91 ns) | | 256 | **0.31 (3266 ns)** — negative scaling | **13.80 (72 ns)** — flat | A warm fts query performs ~2000 cache reads (posting groups + row-id columns across ~350 partitions), so moka's ceiling imposes a hard qps cap. Background `run_pending_tasks` draining was tested and does not help (~+7%): at these rates the 384-slot channel refills past its 64-entry flush point every ~0.13 ms, and the per-read atomics remain. This is the known, still-open moka issue moka-rs/moka#498. ## Change A `CacheBackend` backed by [quick_cache](https://crates.io/crates/quick_cache), whose S3-FIFO/CLOCK-family policy records a hit by setting one reference bit **on the entry itself** — no read log, no reader-triggered housekeeping. Single-flight `get_or_insert` maps to `get_value_or_guard_async`; weighted capacity maps to `Weighter` (u64 weights). **Scope: the session index cache uses this backend; the metadata cache and other general caches stay on moka.** The index cache is where the read-rate wall bites (thousands of cache reads per query on the search path), while moka's richer feature set remains available everywhere else. Custom backends via `Session::with_index_cache_backend` are unaffected. ## Results on a real workload 100M-doc 42-language corpus, 5-term `match_any` (OR), k=100, warm + prewarmed, 320-core node, exact scoring, identical results; same binary (with #7950's chunked search), only the backend differs: | concurrency | moka | quick_cache | |---|---|---| | 16 | 427.6 qps / 37 ms / 22% CPU | 489.3 qps / 33 ms / 25% CPU | | 64 | 220.8 qps / 290 ms / 29% CPU | **1238.7 qps / 52 ms / 79% CPU** | | 128 | 180.7 qps / 710 ms / 47% CPU | **1340.6 qps / 96 ms / 93% CPU** | | 192 | 167.7 qps / 1148 ms / 66% CPU | **1193.9 qps / 161 ms / 86% CPU** | The moka column inverts with concurrency (contention); the quick_cache column peaks at 93% CPU doing scoring work. ## Semantics and scope - Correctness does not depend on eviction order anywhere in lance: entries may be evicted at any time and reload via single-flight (covered by existing fts eviction/reload tests). - Feature parity for what lance actually uses: weighted capacity, async single-flight, iteration, prefix invalidation (implemented via iteration; it has no production call sites today). moka-only features lance does not use: TTL, time-to-idle, eviction listeners. - Eviction policy differs (S3-FIFO vs TinyLFU): comparable hit ratios on published traces, and scan-resistant. Flipping the default should follow a hit-rate comparison under forced memory pressure on representative workloads; until then this is opt-in. ## Tests New backend test exercises roundtrip, single-flight load dedup, weighted-capacity enforcement under overfill, and clear, through the standard `LanceCache` API. `cargo test -p lance-core` green. ## Shard sizing (second commit) quick_cache splits its weight budget evenly across shards with **no cross-shard borrowing**, and an entry heavier than ~its shard's share is silently refused admission (maintainer-confirmed behavior, arthurprs/quick-cache#55). Its default shard count (`available_parallelism × 4`, rounded up to a power of two — 2048 on a 320-core host) is tuned for many-small-entry workloads; index-cache entries are MB-to-hundreds-of-MB posting groups, so small shares risk silent rejection and per-shard imbalance waste. The backend now derives the shard count as `min(cpus / 2, capacity / 4 GiB)`, rounded down to a power of two and clamped to [1, 1024]. The cpu term bounds lock contention (more shards than concurrent threads buys nothing), and the capacity term keeps every shard's share >= 4 GiB so the largest index-cache entries stay admissible — the same shape as RocksDB's block-cache default (`capacity / min_shard_size, capped`). Shard-count sweep (pure get, 50K × 4KB entries, 320-core host, 256 readers) shows contention is a non-issue at low shard counts — even a **single-shard** quick_cache is 10x moka, and throughput plateaus from ~64 shards: | shards | 1 | 4 | 16 | 64 | 128 | 256-2048 | (moka) | |---|---|---|---|---|---|---|---| | Mops/s @ 256 readers | 6.1 | 11.3 | 11.9 | 12.2 | 13.5 | ~13.4-14.2 | 0.63 | On a 320-core host with 960 GiB capacity this yields 128 shards (7.5 GiB share, on the throughput plateau); a 4-core host with 8 GiB yields 2 shards (4 GiB share) — entries up to ~3.9 GiB stay admissible everywhere. A standalone matrix bench across machine profiles (workers 1-320, capacities 2 GiB-1.25 TiB, cache/working-set ratios 0.1-2.0, lance-style 90% 1 MiB + 10% 128 MiB entry mix) confirms every landing point clears the measured demand (~3 Mops of cache gets at 1300 qps FTS) by 2-10x under eviction pressure and 20-100x warm, while moka baselines sit at 0.2-4.7 Mops and degrade as workers grow. End-to-end validation (real 100M-doc 42-language FTS index, prewarmed, c128, 90s/arm, 960 GiB cache): shard counts 1 / 16 / 64 / 128 / 512 / 2048 all measure 1178-1323 qps — inside run-to-run noise — versus ~180 qps on moka. Warm hits take no lock (atomic CLOCK bit), so shard count only affects admission and eviction-phase contention; sizing by admission margin is free. The formula also auto-sized the co-resident 36 GiB metadata cache to 8 shards.
2 participants
Problem
InvertedIndex::bm25_searchspawns one cpu-pool task per partition (~350 per query on a 100-shard index). Under concurrent load this has two costs:index_comparisons(ANALYZE-verified — the comparison count is unchanged; the cost per comparison is what inflates).Change
Partitions are searched in chunks of
LANCE_FTS_SEARCH_CHUNK(default 16):buffer_unordered, so chunk N scores while chunk N+1 loads.A secondary benefit: partitions scored consecutively on one thread observe each other's published shared top-k floor immediately.
LANCE_FTS_SEARCH_CHUNK=1restores the old one-task-per-partition behavior.Results
320-core node, 100M-doc 42-language corpus, 5-term
match_any(OR), tier-100-200 words, k=100, warm + prewarmed, exact scoring, 180s windows, bench returns_rowid+_scoreonly:At c64 throughput is capped by a separate cache-read bottleneck (a follow-up PR swaps the backend); the chunked layout cuts the CPU burned per query ~5x (1.11 → 0.20 core-seconds), which that follow-up then converts into throughput: with the contention-free read path the chunked layout reaches 1340 qps at c128 vs 345 for per-partition tasks.
Tests
cargo test -p lance-index scalar::inverted— 332 passed. The multi-partition tests (41 partitions = 3 chunks) exercise the multi-chunk merge path with exact row_id assertions.