WAL merge leaves physical duplicate rows in the base table; count-based reads over-count

[beinan]

Summary

A WAL merge (RolloutStore::merge_own_shard, folding flushed MemWAL generations into the base table) can leave physically duplicated rows in the base Lance table — the same logical id written twice. The LSM read path de-duplicates by id, so query results are correct. But any code path that counts or scans the base table without de-duplicating over-counts. RolloutStore::observe().row_count is the concrete example we hit, but this is a general property of the merge, not specific to observe.

This is a long-standing issue, independent of any recent write-path changes: it reproduces identically on main.

Reproduction

crates/lance-context-core/tests/wal_merge_generation_cleanup.rs — 30 serial single-row appends against one shard, count-merge every 10 generations. Instrumenting each iteration with the raw count vs. the de-duplicated read path:

iter=8   raw_row_count=9    listed=9    deduped=9    pending_gens=9
iter=9   raw_row_count=10   listed=10   deduped=10   pending_gens=0   <- merge (reclaimed 10)
iter=10  raw_row_count=12   listed=11   deduped=11   pending_gens=1   <- base jumped +2 for +1 append
...
iter=29  raw_row_count=32   listed=30   deduped=30   pending_gens=0   <- merge
final:   raw_row_count=32   listed=30   deduped=30

• list() / get_by_id() (LSM read path) → exactly 30 unique rows. Correct.
• observe().row_count (raw count_rows()) → 32. One physical duplicate is introduced per merge.

Same numbers on main with the unmodified inline-merge add path, so the behavior predates recent work.

Root cause

The base table accumulates a duplicate row each merge because a merged WAL entry is replayed back into a fresh memtable and then re-flushed:

1. Flush stamps the shard manifest's replay_after_wal_entry_position to the WAL position its generation covers (lance memtable/flush.rs, update_manifest: replay_after_wal_entry_position: covered_wal_entry_position). This cursor can lag the just-appended entry by one (off-by-one between "durable in WAL" and "covered by a flushed generation").
2. Merge (crates/lance-context-core/src/rollout_store.rs, merge_own_shard) appends the generations' rows into the base table and drains them from the manifest's flushed_generations, but deliberately does not advance replay_after_wal_entry_position (see the doc comment there: "leaving replay_after_wal_entry_position untouched so a reopened writer does not re-replay already-merged WAL entries"). It also closes the resident writer (claim_epoch would fence it). The underlying WAL log files are never truncated/deleted by a merge — only the manifest's generation list is drained.
3. Reopen: the next append reopens the writer, and replay_memtable_from_wal (lance write.rs, let start_position = manifest.replay_after_wal_entry_position.saturating_add(1)) replays every WAL entry from the cursor to the tail — including the already-merged entry that the lagging cursor did not cover.
4. Re-flush: that replayed row is flushed into a new generation. It now exists in both the base table (from the merge) and the new generation (from the re-flush) → one physical duplicate. count_rows() counts both; the read path dedups by id to one.

Net effect: correctness is preserved by read-time dedup, but the base table's physical row count drifts above the logical row count by ~one per merge.

Why this is broader than observe

Any consumer that treats the base table's physical count_rows() (or a raw, non-deduplicated scan) as the logical row count is affected:

• RolloutStore::observe().row_count = dataset.count_rows(None) + pending_wal_rows(...) (rollout_store.rs ~line 1024–1027; pending_wal_rows also uses count_rows ~line 1741). Reported to the master UI stats table (row_count) and any dashboards/SLOs built on it — they'll read high after merges.
• Any future analytics/export/compaction-sizing logic that scans the base table directly without the LSM dedup would double-count these rows.
• Storage: each duplicate is real bytes in the base table that never get reclaimed (a slow leak of physical rows, distinct from the already-fixed _mem_wal/ directory leak).

The read path itself (list, list_filtered_source, get_by_id, query_sql over the merged view) is not affected — it dedups by id.

Current behavior recap

• Reads: correct (deduped).
• observe().row_count and any raw base-table count: over-count by ~one row per merge.
• Base table: slowly accumulates physical duplicate rows that are never compacted away by id.

Suggested directions (not yet decided)

1. Fix the cursor / replay interaction (root cause). On merge, advance replay_after_wal_entry_position to cover exactly the WAL entries whose generations were folded into the base table (and/or truncate those WAL log entries), so a subsequent reopen does not replay already-merged data. This is the correct fix but touches the merge ↔ MemWAL-replay contract and needs care around crash-safety (the current "leave cursor untouched" choice is deliberate for a reason — a reopen after a partial merge must still recover un-merged tail entries). Likely needs coordination with the lance MemWAL semantics.
2. De-duplicate on the way into the base table during merge. When appending merged generations, drop rows whose id already exists in the base table (or dedup the merged batch set against the base). Bounds the base table to logical row count at the cost of a lookup/anti-join during merge.
3. Make observe() (and any count consumer) dedup-aware. E.g. COUNT(DISTINCT id) or a read-path count. Downside: observe is intentionally a cheap metric; a full dedup scan defeats that. Could be a separate "exact count" API distinct from the cheap estimate.

Options 1/2 fix the underlying physical duplication; option 3 only corrects the reported metric. A combination may be warranted (fix the metric now; schedule the root-cause fix).

Workaround applied

wal_merge_generation_cleanup.rs now asserts correctness via the de-duplicating read path (store.list().len() and unique-id count) instead of observe().row_count, since the read path is the source of truth for "each row present exactly once." This unblocks the test but does not address the underlying physical duplication.