fix(mem_wal): exact PK dedup for LSM vector search and point lookup

[jackye1995]

Duplicate primary keys written into one memtable or across generations leaked through as distinct rows in vector search and point lookup.

Exact PK dedup — replaces the bloom-filter-based FilterStaleExec with an exact pipeline:

• LsmSourceTagExec tags each row with (_memtable_gen, _freshness)
• LsmGlobalPkDedupExec does single-pass cross-source PK dedup keeping the freshest row per PK
• Point lookup adds WithinSourceDedupExec(KeepMaxFreshness) on the active arm

Removes FilterStaleExec, GenerationBloomFilter, and the bloom-filter building from transaction commit.

Post-rerank TakeExec — the planner now accepts a base Dataset reference. After global PK dedup + sort + top-k, a TakeExec materializes any user-projected columns not in the per-source KNN output by fetching from the base dataset via _rowid.

Refine factor plumbing — plan_search() now accepts refine_factor so callers can enable base-table refine (over-fetch k * factor candidates, re-rank with exact distances). Memtable arms use exact HNSW and are unaffected. Exposed in both Python and Java bindings.

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[codecov]

Codecov Report

❌ Patch coverage is 78.92336% with 231 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...rc/dataset/mem_wal/scanner/exec/global_pk_dedup.rs	68.45%	77 Missing and 11 partials ⚠️
...ataset/mem_wal/scanner/exec/within_source_dedup.rs	70.83%	75 Missing and 9 partials ⚠️
...nce/src/dataset/mem_wal/scanner/exec/source_tag.rs	79.14%	45 Missing and 4 partials ⚠️
...lance/src/dataset/mem_wal/scanner/vector_search.rs	95.65%	7 Missing and 1 partial ⚠️
.../lance/src/dataset/mem_wal/scanner/point_lookup.rs	97.89%	0 Missing and 2 partials ⚠️

📢 Thoughts on this report? Let us know!

[hamersaw]

IIUC this algorithm is basically parallelize top-K from each source and then dedup based on freshness. When exploring this elsewhere we noted the bug where a high ranking row is bumped out of the top-K by an un-fresh row and ends up with incorrect results. For a concrete example:

active memtable
    PK=0, _distance=10
    PK=1, _distance=11
    PK=2, _distance=12
flushed memtable (l0 cache) gen 0
    PK=0, _distance=1
    PK=3, _distance=2
    PK=4, _distance=3

If we we a top-K of 2 and dedup on this we have active memtable returning PK 0, 1 and flushed memtable returning PK 0, 3. the dedup results in keys 0, 3 returned. When really, it should be 3, 4 -- both from flushed memtable.

The various mitigations we discussed were:
(1) overfetching each source: we can have some bound that overfetches - don't love it.
(2) incremental refill: if a lower tier source has keys that are update we re-query it to ensure top-K non-updated keys - don't love it.

TBH I thought the bloom filter approach was reasonable to stop duplicates between sources. My thought was that within a source can we simply add a deletion vector to the flushed memtable (l0 cache) on write so that the read tooling automatically removes duplicates without having to rebuild indexes on write.

[jackye1995]

Good catch — this is a real limitation of per-source KNN with post-merge dedup. I looked at how other systems handle this:

Existing approaches:

System	Architecture	Solution
Elasticsearch/Lucene	Per-segment HNSW	Live docs bitmap passed into KNN search (stale docs skipped during traversal) + 2*k overfetch per segment
Milvus	Per-segment KNN	Deletion bitmap applied before per-segment KNN
Qdrant	Per-segment HNSW	Deletion bitmap during search + version-based dedup
Weaviate, Vespa	Single unified HNSW	Problem doesn't exist — graph mutated on every write

All production systems either use a unified index (heavy write cost) or push deletion state to old segments at write time (light write cost, but breaks pure LSM). None do pure read-time resolution.

Two mitigations we should support:

(1) Overfetch (simple, good for latency-sensitive cases)

Same approach as Elasticsearch: each source fetches k * overfetch_factor candidates instead of k. After cross-source dedup, take global top-k. The refine_factor plumbing already exists on plan_search() and could serve double duty here, or we add a separate overfetch_factor parameter. A default of 2x (matching ES) handles most cases. This is a heuristic — not a guarantee — but it's cheap and parallel.

(2) Pre-search bloom filter exclusion node (most accurate)

This is a new idea that's strictly better than the old FilterStaleExec. The old node was a post-search filter — it sat after the union and filtered stale rows from the already-merged per-source top-k results. By that point, displacement had already happened (the stale row had already knocked a valid candidate out of a source's top-k).

The new approach flips the bloom filter to pre-search: before each source runs its KNN, the bloom filters from all newer generations are combined into an exclusion set and passed into the KNN search. During graph traversal (HNSW or IVF), any candidate whose PK matches the exclusion bloom filter is skipped — it never occupies a top-k slot, so no displacement occurs.

Key properties:

• Parallel: bloom filters are precomputed per-generation metadata (built at flush time, stored in manifest), so all sources can be searched concurrently — no sequential dependency
• No write-path mutation: pure read-time, preserves our LSM property of never touching old generations
• More accurate than old FilterStaleExec: old node filtered post-merge (displacement already happened), new node filters pre-search (displacement prevented)
• Same bloom filter tradeoff: ~0.1% false positive rate may skip a small number of valid candidates, same as before

The implementation would pass a bloom-filter-backed predicate into each source's fast_search() as a row-level exclusion filter during index traversal.

[hamersaw]

Lots of great information here, thanks for doing the dive. My thoughts:

• write path mutation is obviously a non-solution.
• using a bloom filter for pre-search would need to modify all of our search algorithms ... which is something we don't want.
• if we go with overfetching what happens if we still shadow some valid PKs? Do we just continue to fetch until it works?

[jackye1995]

MemWAL LSM Vector Search & Point Lookup Benchmark Results

Benchmarked the new exact PK dedup pipeline across LSM levels with two storage backends (S3 and local NVMe) and varying memtable configurations.

Setup: 1M-row base table with IVF-RQ index (256 partitions, 8-bit), 384-dim vectors. Each run has 1 active memtable at 50% capacity plus 0/1/2 flushed generations on disk.

Vector Search (k=10, nprobes=20)

Local NVMe (m6id.8xlarge)

Memtable Size	Flushed Layers	p50 (ms)	p95 (ms)	QPS
baseline	-	7.6	9.1	129
100k	0	8.1	9.8	121
100k	1	8.1	9.9	122
100k	2	7.9	9.9	124
500k	0	8.1	9.8	123
500k	1	7.9	9.4	124
500k	2	7.6	9.3	129
1M	0	7.9	9.5	124
1M	1	8.2	12.4	115
1M	2	8.6	11.2	116

S3 (m7i.12xlarge)

Memtable Size	Flushed Layers	p50 (ms)	p95 (ms)	QPS
baseline	-	79	211	9.5
100k	0	149	330	6.1
100k	1	101	255	8.7
100k	2	125	272	7.6
500k	0	107	236	8.5
500k	1	166	592	4.6
500k	2	142	482	5.1
1M	0	149	348	6.1
1M	1	122	367	6.8
1M	2	122	371	6.7

Observations: On NVMe, latency is stable (~8ms p50) regardless of memtable count or flushed layers — the LSM merge adds negligible overhead on local disk. On S3, latency is 10-20x higher (80-166ms p50) dominated by network round-trips, with tail latency up to 592ms when flushed layers are added.

Point Lookup

Local NVMe

MT Size	Flushed	Base p50 (ms)	Flushed p50 (ms)	Active p50 (ms)
100k	0	2.6	-	2.5
100k	2	3.7	1.4	1.5
500k	0	3.4	-	0.9
500k	2	5.1	2.8	2.9
1M	0	4.2	-	1.6
1M	2	7.2	4.7	4.6

S3

MT Size	Flushed	Base p50 (ms)	Flushed p50 (ms)	Active p50 (ms)
100k	0	66	-	38
100k	2	264	257	250
500k	0	65	-	0.7
500k	2	274	290	274
1M	0	69	-	1.2
1M	2	297	319	311

Observations: Active memtable lookups are fastest when data is warm in memory (sub-1ms on NVMe, sub-2ms on S3 with large memtables). On NVMe, flushed layer reads are fast (1.4-4.7ms) since flushed datasets are small. On S3, adding flushed layers causes a dramatic slowdown — from ~66ms to ~264-319ms (4-5x) — because each flushed generation is probed over the network.

Follow-ups

1. Recall verification with real embeddings

These benchmarks used synthetic clustered embeddings to isolate performance characteristics. To get meaningful recall@k measurements, we need to run against a real embedding dataset. The key question is whether the exact PK dedup pipeline preserves recall compared to a single-table baseline when memtable HNSW results are merged with IVF-RQ base results across generations.

2. Parallel per-generation probing

The current LSM scan probes flushed generations sequentially. This is the primary reason S3 point lookup latency jumps 4-5x when flushed layers are added — each generation requires its own network round-trip, and the cost is additive. When the key lives in the base table (the common case), every flushed generation must be probed and miss before the base is reached. Parallelizing probes (fire all generations concurrently, cancel on first hit for point lookup, merge for vector search) would reduce latency from sum(per_gen_latency) to max(per_gen_latency). The vector search UnionExec already supports parallel partitions, so the main work is in the point lookup path.

[touch-of-grey]

As discussed offline, I'm splitting this up into focused PRs so we can push the progress forward quickly. I need the benchmark template from this PR to reuse for FTS benchmarking, so I'd like to land the pieces incrementally:

• fix(mem_wal): dedupe duplicate primary keys in LSM point lookup #6880 — fix(mem_wal): exact PK dedup for LSM point lookup
• fix(mem_wal): exact PK dedup for LSM vector search #6881 — fix(mem_wal): exact PK dedup for LSM vector search
• benchmark refactor (the reusable CLI/JSON bench template) to follow, stacked on fix(mem_wal): exact PK dedup for LSM vector search #6881 since it depends on the new vector-search API (with_dataset / refine_factor)

Each split PR carries the relevant commits with Co-Authored-By: @jackye1995 and links back here.

[hamersaw]

Thinking about this the distributed top-K could have a stale reads problem as well. Where we update a row that doesn't fit the filter, return the old row, and it's not deduped because the new one isn't returned in an upstream set. This is a tough problem.